feat: protocol negotiation

CHANGELOG.md

@@ -6,6 +6,25 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic
 Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [Unreleased]
+
+### Added
+
+- Client API: protocol version negotiation. `Config::protocol_version`
+  selects the version to advertise in the startup message, defaulting to 3.0.
+  Following libpq, a test version `3.9999` is available to request the newest
+  minor version a server supports: the client handles the server's
+  `NegotiateProtocolVersion` response (both the full 32-bit version form used
+  by PostgreSQL 18+ and the minor-only form used by older servers) and adopts
+  the negotiated version for the rest of the connection.
+
+### Fixed
+
+- Client API: backend messages are now decoded with the rules of the protocol
+  version the client actually advertised, instead of always 3.2. Previously a
+  4-byte protocol 3.0 cancel key was decoded as `SecretKey::Bytes` instead of
+  `SecretKey::I32`.
+
 ## [0.40.7] - 2026-08-12
 
 ### Fixed

src/api/client/auth.rs

@@ -9,10 +9,10 @@ use crate::api::auth::sasl::scram::ScramClientAuth;
 use crate::error::{ErrorInfo, PgWireClientError, PgWireClientResult, PgWireResult};
 use crate::messages::response::ReadyForQuery;
 use crate::messages::startup::{
-    Authentication, BackendKeyData, ParameterStatus, Password, PasswordMessageFamily,
-    SASLInitialResponse, SASLResponse, SecretKey, Startup,
+    Authentication, BackendKeyData, NegotiateProtocolVersion, ParameterStatus, Password,
+    PasswordMessageFamily, SASLInitialResponse, SASLResponse, SecretKey, Startup,
 };
-use crate::messages::{PgWireBackendMessage, PgWireFrontendMessage};
+use crate::messages::{PgWireBackendMessage, PgWireFrontendMessage, ProtocolVersion};
 
 use super::{ClientInfo, ReadyState, ServerInformation};
 
@@ -43,6 +43,10 @@ pub trait StartupHandler: Send {
             PgWireBackendMessage::Authentication(authentication) => {
                 self.on_authentication(client, authentication).await?;
             }
+            PgWireBackendMessage::NegotiateProtocolVersion(negotiation) => {
+                self.on_negotiate_protocol_version(client, negotiation)
+                    .await?;
+            }
             PgWireBackendMessage::ParameterStatus(parameter_status) => {
                 self.on_parameter_status(client, parameter_status).await?;
             }
@@ -78,6 +82,34 @@ pub trait StartupHandler: Send {
             + Send,
         PgWireClientError: From<<C as Sink<PgWireFrontendMessage>>::Error>;
 
+    /// Handle a `NegotiateProtocolVersion` message from the server.
+    ///
+    /// The default implementation adopts the negotiated version for the rest
+    /// of the connection. Both the full 32-bit version number used by
+    /// PostgreSQL 18+ and pgwire, and the minor-only form used by older
+    /// servers, are understood. Startup parameters reported as unrecognized
+    /// by the server are ignored, matching libpq's behavior for non-`_pq_`
+    /// options.
+    async fn on_negotiate_protocol_version<C>(
+        &mut self,
+        client: &mut C,
+        message: NegotiateProtocolVersion,
+    ) -> PgWireClientResult<()>
+    where
+        C: ClientInfo + Sink<PgWireFrontendMessage> + Unpin + Send,
+        PgWireClientError: From<<C as Sink<PgWireFrontendMessage>>::Error>,
+    {
+        match negotiated_version(&message) {
+            Some(version) => {
+                client.set_protocol_version(version);
+                Ok(())
+            }
+            None => Err(PgWireClientError::UnexpectedMessage(Box::new(
+                PgWireBackendMessage::NegotiateProtocolVersion(message),
+            ))),
+        }
+    }
+
     /// Handle a parameter status message from the server.
     async fn on_parameter_status<C>(
         &mut self,
@@ -127,11 +159,17 @@ impl StartupHandler for DefaultStartupHandler {
         C: ClientInfo + Sink<PgWireFrontendMessage> + Unpin + Send,
         PgWireClientError: From<<C as Sink<PgWireFrontendMessage>>::Error>,
     {
-        // TODO: customize protocol version
         let mut startup = Startup::new();
 
         let config = client.config();
 
+        // Advertise the configured protocol version. The connection decodes
+        // with the rules of the advertised version until the server lowers it
+        // via NegotiateProtocolVersion.
+        let (major, minor) = config.get_protocol_version().version_number();
+        startup.protocol_number_major = major;
+        startup.protocol_number_minor = minor;
+
         if let Some(application_name) = &config.application_name {
             startup
                 .parameters
@@ -321,3 +359,107 @@ where
     };
     auth_client.verify_server_final(&message)
 }
+
+/// Interpret the version number reported by a `NegotiateProtocolVersion`
+/// message and return the protocol version to use for the rest of the
+/// connection.
+///
+/// This is the client-side counterpart of the server's
+/// `api::auth::protocol_negotiation`. Two wire dialects exist:
+///
+/// - PostgreSQL 18+ and pgwire report the **full 32-bit protocol version
+///   number** (e.g. `196610` for 3.2), which is always greater than `65535`,
+/// - older servers report only the **minor version** (e.g. `2`), leaving the
+///   major version unchanged from the client's request.
+///
+/// Returns `None` if the reported version cannot be mapped to a version this
+/// crate supports (e.g. a newer major version).
+fn negotiated_version(message: &NegotiateProtocolVersion) -> Option<ProtocolVersion> {
+    let value = message.newest_minor_protocol;
+    if value < 0 {
+        return None;
+    }
+
+    let (major, minor) = if value > i32::from(u16::MAX) {
+        // full 32-bit protocol version number
+        (((value >> 16) & 0xFFFF) as u16, (value & 0xFFFF) as u16)
+    } else {
+        // minor-only form, the major version is unchanged
+        (3, value as u16)
+    };
+
+    match ProtocolVersion::from_version_number(major, minor) {
+        Some(version) => Some(version),
+        // A minor version we don't have a variant for: protocol 3.2 is what
+        // changed the wire formats we care about (secret keys), so an unknown
+        // minor below 2 behaves like 3.0, and one at or above 2 falls back to
+        // our newest 3.x.
+        None if major == 3 && minor >= 2 => Some(ProtocolVersion::PROTOCOL3_2),
+        None if major == 3 => Some(ProtocolVersion::PROTOCOL3_0),
+        None => None,
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_negotiated_version() {
+        // The same version can arrive in two wire dialects, and both forms
+        // must negotiate to the same version:
+        //
+        // - the full 32-bit protocol version number (`i32::from(version)`),
+        //   used by PostgreSQL 18+ and current pgwire,
+        // - the minor version alone, used by older servers, which leaves the
+        //   major version unchanged from the client's request.
+        //
+        // Realistic on the wire: full 3.2 (PostgreSQL 18+), bare 0 (older
+        // PostgreSQL, which only supports 3.0) and bare 2 (pgwire before the
+        // #439 fix sent the minor alone). Full-form 3.0 never occurs, but is
+        // unambiguous.
+        for (full_form, minor_form, expected) in [
+            (
+                i32::from(ProtocolVersion::PROTOCOL3_2),
+                2,
+                ProtocolVersion::PROTOCOL3_2,
+            ),
+            (
+                i32::from(ProtocolVersion::PROTOCOL3_0),
+                0,
+                ProtocolVersion::PROTOCOL3_0,
+            ),
+        ] {
+            assert_eq!(
+                negotiated_version(&NegotiateProtocolVersion::new(full_form, vec![])),
+                Some(expected)
+            );
+            assert_eq!(
+                negotiated_version(&NegotiateProtocolVersion::new(minor_form, vec![])),
+                Some(expected)
+            );
+        }
+
+        // Unknown minors: below 2 behaves like 3.0, at or above 2 falls back
+        // to our newest 3.x
+        assert_eq!(
+            negotiated_version(&NegotiateProtocolVersion::new(1, vec![])),
+            Some(ProtocolVersion::PROTOCOL3_0)
+        );
+        assert_eq!(
+            negotiated_version(&NegotiateProtocolVersion::new(5, vec![])),
+            Some(ProtocolVersion::PROTOCOL3_2)
+        );
+
+        // An unknown major version cannot be mapped
+        assert_eq!(
+            negotiated_version(&NegotiateProtocolVersion::new((4 << 16) | 2, vec![])),
+            None
+        );
+        // A negative value is invalid
+        assert_eq!(
+            negotiated_version(&NegotiateProtocolVersion::new(-1, vec![])),
+            None
+        );
+    }
+}

src/api/client/config.rs

@@ -17,6 +17,7 @@ use std::time::Duration;
 use std::{fmt, iter, mem, str};
 
 use crate::error::PgWireClientError;
+use crate::messages::ProtocolVersion;
 
 /// Properties required of a session.
 #[derive(Debug, Copy, Clone, PartialEq, Eq)]
@@ -222,6 +223,7 @@ pub struct Config {
     pub(crate) target_session_attrs: TargetSessionAttrs,
     pub(crate) channel_binding: ChannelBinding,
     pub(crate) load_balance_hosts: LoadBalanceHosts,
+    pub(crate) protocol_version: ProtocolVersion,
 }
 
 impl Default for Config {
@@ -256,6 +258,7 @@ impl Config {
             target_session_attrs: TargetSessionAttrs::Any,
             channel_binding: ChannelBinding::Prefer,
             load_balance_hosts: LoadBalanceHosts::Disable,
+            protocol_version: ProtocolVersion::PROTOCOL3_0,
         }
     }
 
@@ -551,6 +554,25 @@ impl Config {
         self.load_balance_hosts
     }
 
+    /// Sets the protocol version to advertise in the startup message.
+    ///
+    /// Defaults to `PROTOCOL3_0`.
+    ///
+    /// Following libpq, you can set [`ProtocolVersion::PROTOCOL3_9999`] to
+    /// request the newest minor version the server supports: the server
+    /// replies with `NegotiateProtocolVersion` and the connection proceeds at
+    /// the negotiated version. This is useful for exercising the negotiation
+    /// path against any server.
+    pub fn protocol_version(&mut self, protocol_version: ProtocolVersion) -> &mut Config {
+        self.protocol_version = protocol_version;
+        self
+    }
+
+    /// Gets the protocol version to advertise in the startup message.
+    pub fn get_protocol_version(&self) -> ProtocolVersion {
+        self.protocol_version
+    }
+
     fn param(&mut self, key: &str, value: &str) -> Result<(), PgWireClientError> {
         match key {
             "user" => {
@@ -749,6 +771,7 @@ impl fmt::Debug for Config {
         config_dbg
             .field("target_session_attrs", &self.target_session_attrs)
             .field("channel_binding", &self.channel_binding)
+            .field("protocol_version", &self.protocol_version)
             .finish()
     }
 }

src/api/client/mod.rs

@@ -31,6 +31,15 @@ pub trait ClientInfo {
     /// Returns client protocol version
     fn protocol_version(&self) -> ProtocolVersion;
 
+    /// Sets the protocol version in effect for this connection.
+    ///
+    /// Custom [`StartupHandler`](auth::StartupHandler) implementations should
+    /// call this with the version they advertise in the `Startup` message, so
+    /// that subsequent backend messages are decoded with the rules of that
+    /// version. The default startup handler and the negotiation flow handle
+    /// this automatically.
+    fn set_protocol_version(&mut self, version: ProtocolVersion);
+
     // TODO: transaction state
 }
 

src/api/mod.rs

@@ -347,7 +347,7 @@ impl PidSecretKeyGenerator for RandomPidSecretKeyGenerator {
         let pid = self.next_pid.fetch_add(1, Ordering::Relaxed);
         let secret_key = match client.protocol_version() {
             ProtocolVersion::PROTOCOL3_0 => SecretKey::I32(rand::random::<i32>()),
-            ProtocolVersion::PROTOCOL3_2 => {
+            ProtocolVersion::PROTOCOL3_2 | ProtocolVersion::PROTOCOL3_9999 => {
                 let mut bytes = vec![0u8; 32];
                 rand::fill(&mut bytes);
                 SecretKey::Bytes(bytes.into())

src/messages/mod.rs

@@ -17,6 +17,19 @@ pub enum ProtocolVersion {
     #[default]
     /// Protocol version 3.2
     PROTOCOL3_2,
+    /// Protocol version 3.9999
+    ///
+    /// This is not a real protocol version. Following libpq, a client can
+    /// request this version in its startup message to ask for the newest
+    /// minor version the server supports: a compliant server replies with
+    /// `NegotiateProtocolVersion` and the connection proceeds at the
+    /// negotiated version. This exercises the negotiation path against any
+    /// server.
+    ///
+    /// Note that [`ProtocolVersion::from_version_number`] deliberately does
+    /// not map to this variant: a server must never accept 3.9999 as-is, it
+    /// always negotiates it down to a supported version.
+    PROTOCOL3_9999,
 }
 
 impl ProtocolVersion {
@@ -25,6 +38,7 @@ impl ProtocolVersion {
         match &self {
             Self::PROTOCOL3_0 => (3, 0),
             Self::PROTOCOL3_2 => (3, 2),
+            Self::PROTOCOL3_9999 => (3, 9999),
         }
     }
 
@@ -1028,6 +1042,16 @@ mod test {
     fn test_protocol_version() {
         assert_eq!(196608i32, i32::from(ProtocolVersion::PROTOCOL3_0));
         assert_eq!(196610i32, i32::from(ProtocolVersion::PROTOCOL3_2));
+        assert_eq!(206607i32, i32::from(ProtocolVersion::PROTOCOL3_9999));
+        // 3.9999 is encodable (a client requests it to trigger negotiation)
+        // but is never accepted as a supported version by a server.
+        assert_eq!(
+            None,
+            ProtocolVersion::from_version_number(
+                ProtocolVersion::PROTOCOL3_9999.version_number().0,
+                ProtocolVersion::PROTOCOL3_9999.version_number().1
+            )
+        );
     }
 
     // A count read as signed (0xffff -> -1 -> usize::MAX) must become a decode

src/messages/startup.rs

@@ -505,7 +505,9 @@ impl SecretKey {
         Self::validate_bytes_len(data_len)?;
 
         match ctx.protocol_version {
-            ProtocolVersion::PROTOCOL3_2 => Ok(SecretKey::Bytes(buf.split_to(data_len).freeze())),
+            ProtocolVersion::PROTOCOL3_2 | ProtocolVersion::PROTOCOL3_9999 => {
+                Ok(SecretKey::Bytes(buf.split_to(data_len).freeze()))
+            }
             ProtocolVersion::PROTOCOL3_0 => Ok(SecretKey::I32(buf.get_i32())),
         }
     }
@@ -776,6 +778,13 @@ impl Message for SASLResponse {
 #[non_exhaustive]
 #[derive(PartialEq, Eq, Debug, new)]
 pub struct NegotiateProtocolVersion {
+    /// Version number reported by the server.
+    ///
+    /// PostgreSQL 18+ (protocol 3.2) and pgwire send the **full 32-bit
+    /// protocol version number** here (e.g. `196610` for 3.2), while older
+    /// servers send only the **minor version** (e.g. `2`). The client-side
+    /// negotiation logic in `pgwire::api::client::auth` understands both
+    /// forms.
     pub newest_minor_protocol: i32,
     pub unsupported_options: Vec<String>,
 }