diff --git a/protocols/turnloop-postgres/README.md b/protocols/turnloop-postgres/README.md index f73ded1..b0fe6b2 100644 --- a/protocols/turnloop-postgres/README.md +++ b/protocols/turnloop-postgres/README.md @@ -48,7 +48,10 @@ and RSA-PSS are supported; MD5/SHA-1 signatures use SHA-256. Unsupported algorit 1. Construct `Connection::new(Config { .. })`. The host resolves the address, connects TCP/Unix transport and supplies any absolute connection deadline. -2. Transmit `output()`. Acknowledge **only actually written bytes** with + **Construction already queues output**: `output()` holds the StartupMessage + (or the SSLRequest when `ssl` is `Prefer`/`Require`) before any transport + exists. Do not assume it starts empty; step 2 must run first once connected. +2. Transmit `output()`, starting with those bytes. Acknowledge **only actually written bytes** with `consume_output(n)`. Retain the borrow until write completion, or copy into a reusable host write buffer; never hold a raw pointer while mutating the core. 3. Feed plaintext with `receive(bytes)`, then repeatedly pull `next_event()` until @@ -60,7 +63,10 @@ and RSA-PSS are supported; MD5/SHA-1 signatures use SHA-256. Unsupported algorit `tls_established()` acknowledges TLS without binding data. TLS records never go into `receive`. Prefer mode can fall back after `N`; Require cannot. 5. For `ScramNeeded`, construct the reexported upstream `ScramSha256` **in the - host** (its constructor reads entropy), then call `start_scram`. PLUS needs + host** (its constructor reads entropy) from `connection.config().password`, + then call `start_scram`. The host needs no second copy of the password. Until it + does, `next_event()` returns `Error::State` rather than `Ok(None)`, so an + ignored `ScramNeeded` fails visibly instead of stalling. PLUS needs `ChannelBinding::tls_server_end_point` containing the certificate digest defined in RFC 5929. For plain SCRAM use `ChannelBinding::unsupported()`. The core checks the mechanism/binding selection and @@ -68,13 +74,21 @@ and RSA-PSS are supported; MD5/SHA-1 signatures use SHA-256. Unsupported algorit 6. Schedule `next_timeout()` in the host. Call `handle_timeout(now)` with supplied monotonic time. No method obtains time implicitly. 7. On transport EOF, TLS failure, or an error from `receive`/`next_event`, call - `abort(error)` and drain events. Do not resume parsing after a protocol error. + `abort(error)` and drain events. For a transport failure pass + `Error::transport(&io_error)` (or `Error::Transport(Some(TransportFailure::new(kind, + code, message)))`): the host's kind, error code and text (for example + `connect ECONNREFUSED`) reach every `Outcome::Aborted` and the `Closed` reason. + `Error::Transport(None)` keeps pg's generic message. Do not resume parsing after a protocol error. Each accepted token yields one terminal `Completed`; `Error` and command tags are informational and must not separately settle a promise. Aborting drains all tokens, then emits one `Closed`. Rejected command calls accept no token. Every event borrows storage until the next mutable call. Materialize JS rows, -errors, notifications and fields before advancing. ParameterStatus events allow +errors, notifications and fields before advancing, or retain them with +`Event::into_owned()`: the resulting `OwnedEvent` holds `OwnedRow`, `OwnedFields` +and `OwnedServerError` (also available from `Row`/`Fields`/`ServerError::into_owned`), +each one copy of its wire bytes. `OwnedEvent::as_event()` lends it back as the +borrowed `Event`, so one handler serves both forms. ParameterStatus events allow hosts to retain whatever session parameters they need; the core does not copy an unbounded parameter dictionary. No event calls a host callback itself. @@ -118,7 +132,7 @@ The core has no JavaScript objects; the adapter applies these policies. | json/jsonb | JSON text (jsonb version removed) | host JSON.parse | | uuid | UUID text (binary formatted canonically) | string | | arrays of these | nested Array; NULL preserved | nested JS arrays | -| unknown OID | text string or tagged binary Raw | custom parser or string fallback | +| unknown OID | `Unknown { oid, text }` (never `Text`) or tagged binary `Raw` | custom parser or string fallback | JS Date milliseconds discard microseconds; the host must handle timezone, BC dates and infinity. Binary temporal values retain PostgreSQL's 2000 epoch and diff --git a/protocols/turnloop-postgres/src/client.rs b/protocols/turnloop-postgres/src/client.rs index 8c00d1e..e5358df 100644 --- a/protocols/turnloop-postgres/src/client.rs +++ b/protocols/turnloop-postgres/src/client.rs @@ -103,7 +103,6 @@ impl Client { binding: Option<&[u8]>, at: Instant, ) -> io::Result { - let password = config.password.clone(); let mut driver = AsyncConnection::new( Transport::Plain(stream), crate::Connection::new(config).map_err(io::Error::other)?, @@ -164,9 +163,10 @@ impl Client { } else { ChannelBinding::unsupported() }; + let scram = ScramSha256::new(&driver.core().config().password, channel); driver .core_mut() - .start_scram(ScramSha256::new(&password, channel)) + .start_scram(scram) .map_err(io::Error::other)?; } if ready { diff --git a/protocols/turnloop-postgres/src/lib.rs b/protocols/turnloop-postgres/src/lib.rs index 435e5ea..46ebeb4 100644 --- a/protocols/turnloop-postgres/src/lib.rs +++ b/protocols/turnloop-postgres/src/lib.rs @@ -28,7 +28,9 @@ pub mod pool; pub mod types; mod wire; use wire::Cursor; -pub use wire::{ConnectionFailure, Field, Fields, Row, ServerError}; +pub use wire::{ + ConnectionFailure, Field, Fields, OwnedFields, OwnedRow, OwnedServerError, Row, ServerError, +}; #[cfg(all(target_arch = "wasm32", target_os = "unknown"))] mod host_time; @@ -45,18 +47,29 @@ pub enum Error { State(&'static str), Limit, Timeout, - Transport, + /// The transport failed. The host may attach its own diagnostic (for + /// example `ECONNREFUSED`); it reaches every `Outcome::Aborted` and `Closed`. + Transport(Option), /// The server terminated the session; preserves SQLSTATE and every field. ConnectionAborted(ConnectionFailure), Cancelled, } +impl Error { + /// A transport failure carrying the host's I/O error kind, OS code and text. + pub fn transport(error: &std::io::Error) -> Self { + Self::Transport(Some(error.into())) + } +} impl fmt::Display for Error { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { match self { Self::Protocol(s) | Self::State(s) => f.write_str(s), Self::Limit => f.write_str("protocol buffer limit exceeded"), Self::Timeout => f.write_str("Connection terminated due to timeout"), - Self::Transport => f.write_str("Connection terminated unexpectedly"), + Self::Transport(None) => f.write_str("Connection terminated unexpectedly"), + Self::Transport(Some(failure)) => { + write!(f, "Connection terminated unexpectedly: {failure}") + } Self::ConnectionAborted(error) => write!(f, "Connection aborted: {error}"), Self::Cancelled => f.write_str("Connection terminated"), } @@ -65,14 +78,58 @@ impl fmt::Display for Error { impl std::error::Error for Error {} impl From for std::io::Error { fn from(error: Error) -> Self { - let kind = match error { - Error::ConnectionAborted(_) | Error::Transport => std::io::ErrorKind::ConnectionAborted, + let kind = match &error { + Error::Transport(Some(failure)) => failure.kind(), + Error::ConnectionAborted(_) | Error::Transport(None) => { + std::io::ErrorKind::ConnectionAborted + } Error::Timeout => std::io::ErrorKind::TimedOut, _ => std::io::ErrorKind::Other, }; Self::new(kind, error) } } +/// Host-supplied transport diagnostic. Clones share one copy of the message, +/// so aborting a pipeline of tokens allocates nothing. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct TransportFailure { + kind: std::io::ErrorKind, + code: Option, + message: std::sync::Arc, +} +impl TransportFailure { + /// `code` is whatever error number the host reports (OS errno, libuv code). + pub fn new( + kind: std::io::ErrorKind, + code: Option, + message: impl Into>, + ) -> Self { + Self { + kind, + code, + message: message.into(), + } + } + pub fn kind(&self) -> std::io::ErrorKind { + self.kind + } + pub fn code(&self) -> Option { + self.code + } + pub fn message(&self) -> &str { + &self.message + } +} +impl From<&std::io::Error> for TransportFailure { + fn from(error: &std::io::Error) -> Self { + Self::new(error.kind(), error.raw_os_error(), error.to_string()) + } +} +impl fmt::Display for TransportFailure { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(&self.message) + } +} impl From for Error { fn from(_: std::io::Error) -> Self { Self::Protocol("invalid PostgreSQL message") @@ -131,7 +188,8 @@ pub enum Outcome { #[derive(Debug)] pub enum Event<'a> { UpgradeTls, - /// Build upstream ScramSha256 outside the core (its constructor reads entropy). + /// Build upstream ScramSha256 outside the core (its constructor reads entropy) + /// from `Connection::config().password`, then call `start_scram`. ScramNeeded { plus: bool, }, @@ -189,6 +247,225 @@ pub enum Event<'a> { reason: Error, }, } +/// [`Event`] with every borrowed payload copied out, for hosts that retain +/// rows, errors or notifications past the next mutable call. `as_event` lends +/// it back in borrowed form, so one host handler can consume either. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum OwnedEvent { + UpgradeTls, + ScramNeeded { + plus: bool, + }, + Connected, + ParameterStatus { + name: String, + value: String, + }, + Fields { + token: Token, + fields: OwnedFields, + }, + Row { + token: Token, + row: OwnedRow, + }, + CommandComplete { + token: Token, + tag: String, + row_count: Option, + }, + Error { + token: Option, + error: OwnedServerError, + }, + Notice(OwnedServerError), + Notification { + process_id: i32, + channel: String, + payload: String, + }, + CopyIn { + token: Token, + binary: bool, + column_formats: Vec, + }, + CopyOut { + token: Token, + binary: bool, + column_formats: Vec, + }, + CopyData { + token: Token, + data: Vec, + }, + CopyDone { + token: Token, + }, + Completed { + token: Token, + outcome: Outcome, + transaction: TransactionStatus, + }, + Closed { + reason: Error, + }, +} +impl Event<'_> { + /// Copy the event out of the connection's buffer. Only borrowed payloads + /// allocate: each string, byte slice, row, field list or diagnostic once. + pub fn into_owned(self) -> OwnedEvent { + match self { + Self::UpgradeTls => OwnedEvent::UpgradeTls, + Self::ScramNeeded { plus } => OwnedEvent::ScramNeeded { plus }, + Self::Connected => OwnedEvent::Connected, + Self::ParameterStatus { name, value } => OwnedEvent::ParameterStatus { + name: name.into(), + value: value.into(), + }, + Self::Fields { token, fields } => OwnedEvent::Fields { + token, + fields: fields.into_owned(), + }, + Self::Row { token, row } => OwnedEvent::Row { + token, + row: row.into_owned(), + }, + Self::CommandComplete { + token, + tag, + row_count, + } => OwnedEvent::CommandComplete { + token, + tag: tag.into(), + row_count, + }, + Self::Error { token, error } => OwnedEvent::Error { + token, + error: error.into_owned(), + }, + Self::Notice(error) => OwnedEvent::Notice(error.into_owned()), + Self::Notification { + process_id, + channel, + payload, + } => OwnedEvent::Notification { + process_id, + channel: channel.into(), + payload: payload.into(), + }, + Self::CopyIn { + token, + binary, + column_formats, + } => OwnedEvent::CopyIn { + token, + binary, + column_formats: column_formats.into(), + }, + Self::CopyOut { + token, + binary, + column_formats, + } => OwnedEvent::CopyOut { + token, + binary, + column_formats: column_formats.into(), + }, + Self::CopyData { token, data } => OwnedEvent::CopyData { + token, + data: data.into(), + }, + Self::CopyDone { token } => OwnedEvent::CopyDone { token }, + Self::Completed { + token, + outcome, + transaction, + } => OwnedEvent::Completed { + token, + outcome, + transaction, + }, + Self::Closed { reason } => OwnedEvent::Closed { reason }, + } + } +} +impl OwnedEvent { + /// Borrow the retained event in the form `next_event` produced it. + pub fn as_event(&self) -> Event<'_> { + match self { + Self::UpgradeTls => Event::UpgradeTls, + Self::ScramNeeded { plus } => Event::ScramNeeded { plus: *plus }, + Self::Connected => Event::Connected, + Self::ParameterStatus { name, value } => Event::ParameterStatus { name, value }, + Self::Fields { token, fields } => Event::Fields { + token: *token, + fields: fields.fields(), + }, + Self::Row { token, row } => Event::Row { + token: *token, + row: row.row(), + }, + Self::CommandComplete { + token, + tag, + row_count, + } => Event::CommandComplete { + token: *token, + tag, + row_count: *row_count, + }, + Self::Error { token, error } => Event::Error { + token: *token, + error: error.server_error(), + }, + Self::Notice(error) => Event::Notice(error.server_error()), + Self::Notification { + process_id, + channel, + payload, + } => Event::Notification { + process_id: *process_id, + channel, + payload, + }, + Self::CopyIn { + token, + binary, + column_formats, + } => Event::CopyIn { + token: *token, + binary: *binary, + column_formats, + }, + Self::CopyOut { + token, + binary, + column_formats, + } => Event::CopyOut { + token: *token, + binary: *binary, + column_formats, + }, + Self::CopyData { token, data } => Event::CopyData { + token: *token, + data, + }, + Self::CopyDone { token } => Event::CopyDone { token: *token }, + Self::Completed { + token, + outcome, + transaction, + } => Event::Completed { + token: *token, + outcome: outcome.clone(), + transaction: *transaction, + }, + Self::Closed { reason } => Event::Closed { + reason: reason.clone(), + }, + } + } +} #[derive(Debug, Clone, Copy, PartialEq, Eq)] enum State { Ssl, @@ -250,6 +527,19 @@ pub struct Connection { reason: Error, } impl Connection { + /// Validate `config` and queue the first message, so `output()` is already + /// non-empty on return: the StartupMessage, or with `SslMode::Prefer`/ + /// `Require` the SSLRequest. Send it as soon as the transport connects; + /// nothing else will produce it. + /// + /// ``` + /// use turnloop_postgres::{Config, Connection, SslMode}; + /// let c = Connection::new(Config::default())?; + /// assert!(c.output().windows(5).any(|w| w == b"user\0"), "StartupMessage"); + /// let tls = Connection::new(Config { ssl: SslMode::Require, ..Config::default() })?; + /// assert_eq!(tls.output(), [0, 0, 0, 8, 4, 210, 22, 47], "SSLRequest"); + /// # Ok::<(), turnloop_postgres::Error>(()) + /// ``` pub fn new(config: Config) -> Result { if config.max_buffer < 1024 || config.max_buffer > i32::MAX as usize @@ -283,7 +573,7 @@ impl Connection { copy_in: false, transaction: TransactionStatus::Idle, key: None, - reason: Error::Transport, + reason: Error::Transport(None), }; if this.config.ssl == SslMode::Disable { this.startup()?; @@ -307,6 +597,11 @@ impl Connection { self.state = State::Auth; Ok(()) } + /// The configuration this connection was built from. On `ScramNeeded`, + /// build `ScramSha256` from `config().password`; the host keeps no copy. + pub fn config(&self) -> &Config { + &self.config + } pub fn output(&self) -> &[u8] { &self.output[self.output_at..] } @@ -626,6 +921,11 @@ impl Connection { /// `Error::ConnectionAborted` immediately and makes the session unusable. /// Continue pulling to drain one Aborted completion per pending token, then /// Closed; a later transport abort cannot replace the server diagnostic. + /// + /// After `ScramNeeded`, pulling again before `start_scram` returns + /// `Error::State` instead of `Ok(None)`: an unanswered SCRAM request is a + /// host error, not idleness. The connection is unchanged, so `start_scram` + /// (or `abort`) is still accepted afterwards. pub fn next_event(&mut self) -> Result>> { if self.state == State::Closing { if let Some(p) = self.pending.pop_front() { @@ -643,7 +943,14 @@ impl Connection { reason: self.reason.clone(), })); } - if matches!(self.state, State::Closed | State::Tls | State::Scram(_)) { + if let State::Scram(_) = self.state { + // Nothing can progress until the host answers ScramNeeded; an + // Ok(None) here would look like "no input yet" and stall silently. + return Err(Error::State( + "SCRAM authentication pending: answer ScramNeeded with start_scram", + )); + } + if matches!(self.state, State::Closed | State::Tls) { return Ok(None); } if self.state == State::Ssl { diff --git a/protocols/turnloop-postgres/src/types.rs b/protocols/turnloop-postgres/src/types.rs index d4e3fd0..226410a 100644 --- a/protocols/turnloop-postgres/src/types.rs +++ b/protocols/turnloop-postgres/src/types.rs @@ -34,6 +34,13 @@ pub enum Value<'a> { oid: u32, bytes: Cow<'a, [u8]>, }, + /// Text-format value of an OID with no codec here (time, interval, inet, + /// oid, xml, enums, ...). Distinct from `Text`, which only text, varchar, + /// bpchar and name produce, so a host can apply its own parser. + Unknown { + oid: u32, + text: Cow<'a, str>, + }, } impl Value<'_> { pub fn into_owned(self) -> Value<'static> { @@ -65,6 +72,10 @@ impl Value<'_> { oid, bytes: Cow::Owned(bytes.into_owned()), }, + Self::Unknown { oid, text } => Value::Unknown { + oid, + text: Cow::Owned(text.into_owned()), + }, } } } @@ -132,7 +143,12 @@ pub fn decode(oid: u32, format: i16, bytes: Option<&[u8]>) -> Result> oid, text: s.into(), }, - _ => Value::Text(s.into()), + // The same text OIDs the binary codec accepts. + 19 | 25 | 1042 | 1043 => Value::Text(s.into()), + _ => Value::Unknown { + oid, + text: s.into(), + }, }); } let mut c = Cursor(b); @@ -488,4 +504,44 @@ mod tests { Value::Array(vec![Value::Int(42), Value::Null]) ); } + #[test] + fn unknown_text_oids_are_not_reported_as_text() { + for oid in [19, 25, 1042, 1043] { + assert_eq!( + decode(oid, 0, Some(b"hello")).unwrap(), + Value::Text("hello".into()), + "OID {oid}" + ); + } + // time, interval, inet, oid and an arbitrary enum/extension OID. + for (oid, text) in [ + (1083, "12:34:56"), + (1186, "1 day"), + (869, "127.0.0.1"), + (26, "16384"), + (91_234, "happy"), + ] { + let value = decode(oid, 0, Some(text.as_bytes())).unwrap(); + assert_eq!( + value, + Value::Unknown { + oid, + text: text.into() + } + ); + assert!(matches!( + value.into_owned(), + Value::Unknown { oid: o, text: Cow::Owned(t) } if o == oid && t == text + )); + } + // Binary unknowns stay uninterpreted bytes. + assert_eq!( + decode(1083, 1, Some(&[0; 8])).unwrap(), + Value::Raw { + oid: 1083, + bytes: Cow::Borrowed(&[0; 8]) + } + ); + assert!(decode(1083, 0, Some(&[0xff])).is_err()); + } } diff --git a/protocols/turnloop-postgres/src/wire.rs b/protocols/turnloop-postgres/src/wire.rs index c67ab30..96bea9e 100644 --- a/protocols/turnloop-postgres/src/wire.rs +++ b/protocols/turnloop-postgres/src/wire.rs @@ -96,6 +96,36 @@ impl<'a> Iterator for Fields<'a> { } } impl ExactSizeIterator for Fields<'_> {} +impl Fields<'_> { + /// Copy the unread field descriptions out of the connection's buffer. + /// One allocation; the result outlives every later mutable call. + pub fn into_owned(self) -> OwnedFields { + OwnedFields { + data: self.cursor.0.into(), + remaining: self.remaining, + } + } +} +/// Owned RowDescription. `fields()` iterates it exactly like the borrowed form. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct OwnedFields { + data: Box<[u8]>, + remaining: u16, +} +impl OwnedFields { + pub fn fields(&self) -> Fields<'_> { + Fields { + cursor: Cursor(&self.data), + remaining: self.remaining, + } + } + pub fn len(&self) -> usize { + self.remaining as usize + } + pub fn is_empty(&self) -> bool { + self.remaining == 0 + } +} /// Borrowed, already validated PostgreSQL row. Iteration never allocates. #[derive(Debug, Clone, Copy)] @@ -137,6 +167,41 @@ impl<'a> Iterator for Row<'a> { } } impl ExactSizeIterator for Row<'_> {} +impl Row<'_> { + /// Copy the unread values out of the connection's buffer in one allocation, + /// so the row can be retained past the next mutable call on the connection. + pub fn into_owned(self) -> OwnedRow { + OwnedRow { + data: self.cursor.0.into(), + remaining: self.remaining, + } + } +} +/// Owned, already validated DataRow. `row()` lends the same borrowed iterator, +/// so one host decoder serves both forms; `get` indexes a column directly. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct OwnedRow { + data: Box<[u8]>, + remaining: u16, +} +impl OwnedRow { + pub fn row(&self) -> Row<'_> { + Row { + cursor: Cursor(&self.data), + remaining: self.remaining, + } + } + /// `None` past the last column; `Some(None)` is SQL NULL. + pub fn get(&self, index: usize) -> Option> { + self.row().nth(index)?.ok() + } + pub fn len(&self) -> usize { + self.remaining as usize + } + pub fn is_empty(&self) -> bool { + self.remaining == 0 + } +} /// ErrorResponse and NoticeResponse retain every field, including future fields. #[derive(Clone, Copy, Debug)] @@ -165,7 +230,30 @@ impl std::fmt::Debug for ConnectionFailure { f.debug_map().entries(self.server_error().fields()).finish() } } +/// Owned ErrorResponse/NoticeResponse with every field, including future ones. +#[derive(Clone, PartialEq, Eq)] +pub struct OwnedServerError(Box<[u8]>); +impl OwnedServerError { + pub fn server_error(&self) -> ServerError<'_> { + ServerError(&self.0) + } +} +impl std::fmt::Display for OwnedServerError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + let error = self.server_error(); + write!(f, "{}: {}", error.code(), error.message()) + } +} +impl std::fmt::Debug for OwnedServerError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_map().entries(self.server_error().fields()).finish() + } +} impl<'a> ServerError<'a> { + /// Copy every field out of the connection's buffer in one allocation. + pub fn into_owned(self) -> OwnedServerError { + OwnedServerError(self.0.into()) + } pub(crate) fn parse(body: &'a [u8]) -> Result { let mut c = Cursor(body); while c.u8()? != 0 { diff --git a/protocols/turnloop-postgres/tests/allocations.rs b/protocols/turnloop-postgres/tests/allocations.rs index b062b49..4223621 100644 --- a/protocols/turnloop-postgres/tests/allocations.rs +++ b/protocols/turnloop-postgres/tests/allocations.rs @@ -140,7 +140,7 @@ fn terminal_diagnostic_allocates_once_and_pipeline_abort_clones_allocate_zero() let diagnostic = diagnostic.expect("owned error"); assert_eq!( allocation::allocations(|| { - c.abort(Error::Transport); + c.abort(Error::Transport(None)); for token in 0..64 { match c.next_event().expect("completion") { Some(Event::Completed { @@ -162,4 +162,44 @@ fn terminal_diagnostic_allocates_once_and_pipeline_abort_clones_allocate_zero() 0 ); assert_eq!(c.pending_count(), 0); + + // A host transport diagnostic is built once; fanning it out to every + // pending token and the close shares that copy. + let mut c = Connection::new(Config::default()).expect("core"); + c.consume_output(c.output().len()).expect("startup sent"); + c.receive(b"R\0\0\0\x08\0\0\0\0Z\0\0\0\x05I").expect("auth"); + assert!(matches!( + c.next_event().expect("connected"), + Some(Event::Connected) + )); + for token in 0..64 { + c.query(token, "SELECT 1", None).expect("pipeline"); + } + let reason = Error::Transport(Some(TransportFailure::new( + std::io::ErrorKind::ConnectionReset, + Some(-104), + "read ECONNRESET", + ))); + assert_eq!( + allocation::allocations(|| { + c.abort(reason.clone()); + for token in 0..64 { + match c.next_event().expect("completion") { + Some(Event::Completed { + token: actual, + outcome: Outcome::Aborted(error), + .. + }) => { + assert_eq!(actual, token); + assert_eq!(error, reason); + } + event => panic!("missing pipeline abort: {event:?}"), + } + } + assert!( + matches!(c.next_event().expect("close"), Some(Event::Closed { reason: r }) if r == reason) + ); + }), + 0 + ); } diff --git a/protocols/turnloop-postgres/tests/protocol.rs b/protocols/turnloop-postgres/tests/protocol.rs index c788aea..39312e8 100644 --- a/protocols/turnloop-postgres/tests/protocol.rs +++ b/protocols/turnloop-postgres/tests/protocol.rs @@ -316,6 +316,7 @@ fn scram_and_plus_verify_server_signature_and_reject_bad_verifier() { SslMode::Disable }, channel_binding_required: plus, + password: b"secret".to_vec(), ..Config::default() }) .expect("fixture operation must succeed"); @@ -342,7 +343,10 @@ fn scram_and_plus_verify_server_signature_and_reject_bad_verifier() { } else { ChannelBinding::unsupported() }; - let scram = ScramSha256::new(b"secret", binding); + // The consumed Config still supplies the password; the host keeps no + // copy. The server signature below is derived from "secret". + assert_eq!(c.config().password, b"secret"); + let scram = ScramSha256::new(&c.config().password, binding); let first = std::str::from_utf8(scram.message()) .expect("fixture operation must succeed") .to_owned(); @@ -584,6 +588,15 @@ fn legacy_tls_acknowledgement_has_no_binding_and_uses_n_gs2_flag() { c.next_event().expect("selection"), Some(Event::ScramNeeded { plus: false }) )); + // A host that ignores ScramNeeded must not see the Ok(None) of an idle + // connection; the error repeats and leaves the request answerable. + for _ in 0..2 { + assert_eq!( + c.next_event().expect_err("unanswered SCRAM request"), + Error::State("SCRAM authentication pending: answer ScramNeeded with start_scram") + ); + } + assert!(c.output().is_empty()); c.start_scram(ScramSha256::new(b"secret", ChannelBinding::unsupported())) .expect("plain SCRAM"); let packet = flush(&mut c); @@ -632,7 +645,7 @@ fn fatal_errors_abort_every_pending_token_and_preserve_diagnostics_after_eof() { assert!(!c.is_ready()); assert!(c.output().is_empty(), "discard unsent pipeline output"); assert!(c.query(99, "SELECT 1", None).is_err()); - c.abort(Error::Transport); // Scripted peer closes immediately after FATAL. + c.abort(Error::Transport(None)); // Scripted peer closes immediately after FATAL. for token in 1..=pending { match c.next_event().expect("terminal completion") { Some(Event::Completed { @@ -707,3 +720,187 @@ fn nonlocalized_severity_controls_statement_error_recovery() { }) )); } + +#[test] +fn owned_events_survive_later_input_and_lend_back_the_borrowed_form() { + let mut c = ready(); + c.query(1, "SELECT answer, note", None).expect("query"); + flush(&mut c); + let mut fields = vec![0, 2]; + for name in [b"answer\0".as_slice(), b"note\0"] { + fields.extend_from_slice(name); + fields.extend_from_slice(&[ + 0, 0, 0, 9, 0, 1, 0, 0, 0, 23, 0, 4, 255, 255, 255, 255, 0, 0, + ]); + } + let first = [ + frame(b'T', &fields), + frame(b'D', &[0, 2, 0, 0, 0, 2, b'4', b'2', 255, 255, 255, 255]), + frame(b'N', b"SNOTICE\0C00000\0Mfirst notice\0\0"), + frame(b'A', b"\0\0\0\x07chan\0hello\0"), + ] + .concat(); + c.receive(&first).expect("first batch"); + let mut retained = Vec::new(); + while let Some(event) = c.next_event().expect("event") { + retained.push(event.into_owned()); + } + assert_eq!(retained.len(), 4); + // The next receive compacts the input buffer over the bytes those events + // borrowed; the owned copies must be unaffected. + let second = [ + frame(b'D', &[0, 2, 0, 0, 0, 2, b'9', b'9', 0, 0, 0, 1, b'x']), + frame(b'C', b"SELECT 2\0"), + frame(b'Z', b"I"), + ] + .concat(); + c.receive(&second).expect("second batch"); + while let Some(event) = c.next_event().expect("event") { + retained.push(event.into_owned()); + } + assert_eq!(retained.len(), 7); + + let OwnedEvent::Fields { token: 1, fields } = &retained[0] else { + panic!("fields: {:?}", retained[0]); + }; + assert_eq!(fields.len(), 2); + let names: Vec<_> = fields + .fields() + .map(|f| { + let f = f.expect("field"); + (f.name.to_owned(), f.table_id, f.data_type_id) + }) + .collect(); + assert_eq!( + names, + [("answer".to_owned(), 9, 23), ("note".to_owned(), 9, 23)] + ); + let OwnedEvent::Row { token: 1, row } = &retained[1] else { + panic!("row: {:?}", retained[1]); + }; + assert_eq!(row.len(), 2); + assert_eq!(row.get(0), Some(Some(b"42".as_slice()))); + assert_eq!(row.get(1), Some(None)); + assert_eq!(row.get(2), None); + let OwnedEvent::Notice(notice) = &retained[2] else { + panic!("notice: {:?}", retained[2]); + }; + assert_eq!(notice.server_error().message(), "first notice"); + assert_eq!(notice.to_string(), "00000: first notice"); + assert_eq!( + retained[3], + OwnedEvent::Notification { + process_id: 7, + channel: "chan".into(), + payload: "hello".into(), + } + ); + let OwnedEvent::Row { + row: second_row, .. + } = &retained[4] + else { + panic!("second row: {:?}", retained[4]); + }; + assert_eq!(second_row.get(0), Some(Some(b"99".as_slice()))); + assert_eq!(second_row.get(1), Some(Some(b"x".as_slice()))); + assert_eq!( + retained[5], + OwnedEvent::CommandComplete { + token: 1, + tag: "SELECT 2".into(), + row_count: Some(2), + } + ); + assert_eq!( + retained[6], + OwnedEvent::Completed { + token: 1, + outcome: Outcome::Success, + transaction: TransactionStatus::Idle, + } + ); + + // A handler written for borrowed events consumes retained ones unchanged. + let Event::Row { token: 1, row } = retained[1].as_event() else { + panic!("borrowed row"); + }; + let values: Vec<_> = row.map(|v| v.expect("value")).collect(); + assert_eq!(values, [Some(b"42".as_slice()), None]); + assert!(matches!( + retained[2].as_event(), + Event::Notice(error) if error.code() == "00000" + )); + assert_eq!(retained[1].as_event().into_owned(), retained[1]); +} + +#[test] +fn host_transport_diagnostic_reaches_every_aborted_completion_and_close() { + let refused = std::io::Error::new( + std::io::ErrorKind::ConnectionRefused, + "connect ECONNREFUSED 127.0.0.1:5432", + ); + let from_io = Error::transport(&refused); + let with_code = Error::Transport(Some(TransportFailure::new( + std::io::ErrorKind::ConnectionReset, + Some(-104), + "read ECONNRESET", + ))); + for (reason, kind, code, message) in [ + ( + from_io, + std::io::ErrorKind::ConnectionRefused, + None, + "connect ECONNREFUSED 127.0.0.1:5432", + ), + ( + with_code, + std::io::ErrorKind::ConnectionReset, + Some(-104), + "read ECONNRESET", + ), + ] { + let mut c = ready(); + for token in 1..=3 { + c.query(token, "SELECT 1", None).expect("queue query"); + } + c.abort(reason.clone()); + for token in 1..=3 { + match c.next_event().expect("terminal completion") { + Some(Event::Completed { + token: actual, + outcome: Outcome::Aborted(Error::Transport(Some(failure))), + .. + }) => { + assert_eq!(actual, token); + assert_eq!(failure.kind(), kind); + assert_eq!(failure.code(), code); + assert_eq!(failure.message(), message); + } + event => panic!("lost transport diagnostic: {event:?}"), + } + } + let Some(Event::Closed { reason: closed }) = c.next_event().expect("close") else { + panic!("missing close"); + }; + assert_eq!(closed, reason); + assert_eq!( + closed.to_string(), + format!("Connection terminated unexpectedly: {message}") + ); + let io = std::io::Error::from(closed); + assert_eq!(io.kind(), kind); + assert!(io.to_string().ends_with(message)); + } + // An OS error keeps its raw code. + let os = std::io::Error::from_raw_os_error(2); + let Error::Transport(Some(failure)) = Error::transport(&os) else { + panic!("transport"); + }; + assert_eq!(failure.code(), Some(2)); + assert_eq!(failure.message(), os.to_string()); + // Without a diagnostic the generic pg message is unchanged. + assert_eq!( + Error::Transport(None).to_string(), + "Connection terminated unexpectedly" + ); +} diff --git a/protocols/turnloop-postgres/tests/server.rs b/protocols/turnloop-postgres/tests/server.rs index 1104824..ad802b0 100644 --- a/protocols/turnloop-postgres/tests/server.rs +++ b/protocols/turnloop-postgres/tests/server.rs @@ -23,7 +23,6 @@ struct Results { struct Driver { core: Connection, io: Transport, - password: Vec, ssl: bool, scram: bool, plus: bool, @@ -40,10 +39,9 @@ impl Driver { } fn connect_to(port: u16, user: &str, ssl: SslMode) -> Self { eprintln!("PostgreSQL connecting as {user}, SSL mode {ssl:?}"); - let password = b"fixture-password".to_vec(); let config = Config { user: user.into(), - password: password.clone(), + password: b"fixture-password".to_vec(), ssl, connect_deadline: Some(Instant::now() + Duration::from_secs(10)), ..Config::default() @@ -52,7 +50,6 @@ impl Driver { let mut d = Self { core, io: Transport::connect(port), - password, ssl: false, scram: false, plus: false, @@ -103,8 +100,9 @@ impl Driver { } else { ChannelBinding::unsupported() }; + let scram = ScramSha256::new(&self.core.config().password, binding); self.core - .start_scram(ScramSha256::new(&self.password, binding)) + .start_scram(scram) .expect("fixture operation must succeed"); self.scram = true; self.plus = plus; @@ -399,7 +397,10 @@ fn all_common_types_in_text_binary_and_arrays() { assert_eq!(text.fields[0].1, oid); let text_value = decode(oid, 0, text.rows[0][0].as_deref()).expect("fixture operation must succeed"); - assert!(!matches!(text_value, Value::Null | Value::Raw { .. })); + assert!(!matches!( + text_value, + Value::Null | Value::Raw { .. } | Value::Unknown { .. } + )); d.core .execute( 2,