diff --git a/protocols/turnloop-mysql/README.md b/protocols/turnloop-mysql/README.md index 3a4491f..8b03647 100644 --- a/protocols/turnloop-mysql/README.md +++ b/protocols/turnloop-mysql/README.md @@ -41,7 +41,10 @@ caching_sha2 RSA authentication obtains fresh OAEP entropy from the TLS provider Construct Config and Connection, connect transport in the host, then feed plaintext into `receive`. Pull `next_event` until None; `Progress` means a control packet was consumed and polling should continue. Write `output`, acknowledging -only successfully written bytes via `consume_output`. Borrowed outputs/events +only successfully written bytes via `consume_output`. When it returns `true`, an +event is ready without further input: call `next_event` immediately. COM_STMT_CLOSE +and COM_QUIT get no server reply, so that acknowledgement is their only wakeup; +a completion-driven host needs no timer for them. Borrowed outputs/events remain valid until the next mutable call. Preserve the output borrow through write completion or copy into a reusable host transport buffer. @@ -58,14 +61,18 @@ Provide absolute deadlines; schedule `next_timeout` and call deadline; Config does not read a clock. On EOF/TLS failure or a parsing error, call `abort(error)` and drain events. A parsing error is terminal; never continue the byte stream after it. Each accepted command yields exactly one Completed; -server Error and result Ok events are informational. Close emits one Closed. +server Error, Ok and Eof events are informational. Close emits one Closed. Command rejection accepts no token. There are no callbacks from this core. -MySQL permits one active command. Busy calls return backpressure; the adapter +MySQL permits one active command. Busy calls return backpressure; +`can_accept()` reports that exact admission decision up front (the command +methods call it themselves), so a host queue never copies the rule. The adapter queues commands in JS submission order. This prevents unsynchronized packet sequence resets. COM_QUERY supports multiple result sets (multiple statements are opt-in); EOF negotiation deliberately selects legacy EOF, which MySQL 9.6 -supports. Result Ok includes affected_rows, last_insert_id, warnings and status. +supports. `Ok` is a real OK packet (affected_rows, last_insert_id, warnings, +status); `Eof` ends a result set's rows and carries only warnings and status, +so a host need not track whether a result set is open to read either one. Prepare emits parameter/column metadata and a Statement ID. Execute accepts mysql_common Value parameters and emits binary rows. Reset/close statements, @@ -94,18 +101,26 @@ Zstd and MariaDB extensions are out of scope. ## mysql2 conversions and hooks Row iterators return borrowed bytes or mysql_common numeric/calendar scalars. +Each value is validated and decoded once, as the iterator reaches it (the last +one also rejects trailing bytes); the first error ends the row. Such an error +is a parsing error: abort the connection. `types::decode` implements a default conversion policy. `Column` retains names, -original names/table/schema, flags, charset, length, decimals and wire type. +original names/table/schema, flags, charset, length, decimals and wire type; +every `Row` also carries its columns' `ColumnTypeInfo` (type, flags, charset, +decimals) via `Row::columns` and `Row::typed`, so a host can implement its own +policy per MySQL column type without reimplementing value decoding. | MySQL type | Default policy / host action | |---|---| | integer, FLOAT, DOUBLE | Number; TINYINT(1) remains numeric | | BIGINT | Number by default; support_big_numbers preserves unsafe-range values as strings; big_number_strings forces strings when enabled | | DECIMAL/NEWDECIMAL | exact string; decimal_numbers opts into f64 | -| DATE/DATETIME/TIMESTAMP | explicit Date request with raw text/calendar components; date_strings formats strings | -| TIME | string, including negative and >24-hour values | +| DATE/DATETIME/TIMESTAMP | explicit Date request with raw text/calendar components; date_strings formats strings; truncate_fraction_to_decimals cuts binary-protocol fractions to the column's decimals like mysql2 | +| TIME | string, including negative and >24-hour values (by column type, even though its charset is 63) | | JSON | explicit host JSON parse request; json_strings returns raw text | -| BLOB/binary charset 63, BIT, geometry | Buffer bytes | +| BIT | Buffer bytes | +| GEOMETRY | explicit Geometry request with the SRID+WKB bytes; the host builds mysql2's objects | +| BLOB/other string types with binary charset 63 | Buffer bytes | | text | UTF-8 string; other character sets require host decoding | | NULL | Null | @@ -114,7 +129,7 @@ behavior), parses JSON and materializes row objects or rowsAsArray tuples. The protocol preserves microseconds; JS Date loses sub-millisecond precision. `typeCast` can inspect Column and RawValue in the event-dispatch layer and invoke `types::decode` for next(). Callback invocation, field.string/buffer single-use -semantics and field.geometry parsing are adapter work, not core callbacks. +semantics and building field.geometry objects are adapter work, not core callbacks. Per-type dateStrings arrays are not implemented (only the boolean option). BIGINT inside JSON is still host policy. This is documented surface support, not a drop-in mysql2 API. diff --git a/protocols/turnloop-mysql/src/asynchronous.rs b/protocols/turnloop-mysql/src/asynchronous.rs index 3b02972..0a74884 100644 --- a/protocols/turnloop-mysql/src/asynchronous.rs +++ b/protocols/turnloop-mysql/src/asynchronous.rs @@ -10,7 +10,9 @@ impl Output for crate::Connection { self.output() } fn consume_output(&mut self, n: usize) -> io::Result<()> { - self.consume_output(n).map_err(io::Error::other) + // The driver polls `event` after every drain, so the "call next_event + // now" signal is already honoured here. + self.consume_output(n).map(drop).map_err(io::Error::other) } } impl SansIo for crate::Connection { diff --git a/protocols/turnloop-mysql/src/lib.rs b/protocols/turnloop-mysql/src/lib.rs index 4e3d869..e155c86 100644 --- a/protocols/turnloop-mysql/src/lib.rs +++ b/protocols/turnloop-mysql/src/lib.rs @@ -140,10 +140,21 @@ pub enum Event<'a> { token: Token, row: Row<'a>, }, + /// A genuine OK packet: a command or one statement of a multi-statement + /// query finished without a result set. `affected_rows`, `last_insert_id` + /// and `info` are meaningful. Ok { token: Token, packet: OkPacket<'a>, }, + /// The EOF packet ending one result set's rows. It carries no row counts, + /// only the warning count and status flags (`SERVER_MORE_RESULTS_EXISTS` + /// announces another result). + Eof { + token: Token, + warnings: u16, + status: StatusFlags, + }, Prepared { token: Token, statement: Statement, @@ -193,6 +204,8 @@ enum CommandKind { Execute, Reset, ChangeUser, + /// COM_STMT_CLOSE for this statement ID; forgotten once its bytes are sent. + CloseStatement(u32), Other, } struct Pending { @@ -270,7 +283,14 @@ impl Connection { pub fn output(&self) -> &[u8] { &self.output[self.output_at..] } - pub fn consume_output(&mut self, n: usize) -> Result<()> { + /// Acknowledge `n` written bytes of `output()`. + /// + /// Returns `true` when an event is now deliverable without further input, + /// in which case the host must call `next_event()` right away. This matters + /// for COM_STMT_CLOSE and COM_QUIT: the server never answers them, so their + /// `Completed` / `Closed` is produced by this acknowledgement itself and no + /// read completion or timer would otherwise wake a completion-driven host. + pub fn consume_output(&mut self, n: usize) -> Result { if n > self.output().len() { return Err(Error::State("invalid output acknowledgement")); } @@ -278,8 +298,21 @@ impl Connection { if self.output_at == self.output.len() { self.output.clear(); self.output_at = 0; + if self.state == State::NoResponse { + // The server holds the statement until these bytes reach it, so + // it is dropped from the bookkeeping only now; a failed flush + // aborts instead and leaves it listed. + if let Some(Pending { + kind: CommandKind::CloseStatement(id), + .. + }) = self.pending + { + self.statements.retain(|s| s.id != id); + } + self.complete(Outcome::Success); + } } - Ok(()) + Ok(self.completion.is_some() || (self.state == State::Closing && self.output.is_empty())) } pub fn receive(&mut self, b: &[u8]) -> Result<()> { if matches!( @@ -403,8 +436,16 @@ impl Connection { self.state = State::Auth; Ok(()) } + /// Whether a command submitted now would be admitted. This is the exact rule + /// every command method applies: the session is authenticated and idle, the + /// previous command's `Completed` has been delivered and all output has been + /// acknowledged. A host queue can gate submission on it instead of copying + /// the preconditions. Unlike `is_ready`, it also requires flushed output. + pub fn can_accept(&self) -> bool { + self.state == State::Ready && self.pending.is_none() && self.output().is_empty() + } fn accept(&self) -> Result<()> { - if self.state != State::Ready || self.pending.is_some() || !self.output().is_empty() { + if !self.can_accept() { Err(Error::State("connection busy or closed")) } else { Ok(()) @@ -537,9 +578,18 @@ impl Connection { self.simple(token, 0x1a, Some(id), CommandKind::Other, State::Header) } pub fn close_statement(&mut self, token: Token, id: u32) -> Result<()> { - self.simple(token, 0x19, Some(id), CommandKind::Other, State::NoResponse)?; - self.statements.retain(|s| s.id != id); - Ok(()) + self.simple( + token, + 0x19, + Some(id), + CommandKind::CloseStatement(id), + State::NoResponse, + ) + } + /// Prepared statements the server still holds for this session. A statement + /// being closed stays listed until its COM_STMT_CLOSE bytes are acknowledged. + pub fn statements(&self) -> &[Statement] { + &self.statements } pub fn change_user( &mut self, @@ -642,9 +692,6 @@ impl Connection { self.state = State::Ready; } pub fn next_event(&mut self) -> Result>> { - if self.state == State::NoResponse && self.output().is_empty() { - self.complete(Outcome::Success); - } if let Some(outcome) = self.completion.take() { let p = self .pending @@ -902,8 +949,9 @@ impl Connection { } State::Rows => { if self.packet[0] == 0xfe && self.packet.len() < 9 { - let ok = parse_eof(&self.packet, self.caps)?; - self.status = ok.status_flags(); + let eof = parse_eof(&self.packet, self.caps)?; + let warnings = eof.warnings(); + self.status = eof.status_flags(); if self .status .contains(StatusFlags::SERVER_MORE_RESULTS_EXISTS) @@ -912,9 +960,10 @@ impl Connection { } else { self.complete(Outcome::Success); } - return Ok(Some(Event::Ok { + return Ok(Some(Event::Eof { token: self.token()?, - packet: parse_eof(&self.packet, self.caps)?, + warnings, + status: self.status, })); } return Ok(Some(Event::Row { diff --git a/protocols/turnloop-mysql/src/types.rs b/protocols/turnloop-mysql/src/types.rs index a2bf467..d7e1881 100644 --- a/protocols/turnloop-mysql/src/types.rs +++ b/protocols/turnloop-mysql/src/types.rs @@ -9,6 +9,12 @@ pub struct Options { pub big_number_strings: bool, pub date_strings: bool, pub json_strings: bool, + /// mysql2's binary-protocol DATETIME/TIMESTAMP string policy: cut the + /// fractional seconds of a formatted `date_strings` value to the column's + /// declared `decimals` (DATETIME(3) gives `.123`, not `.123000`). Without + /// it all six digits are kept. Text-protocol values already arrive with the + /// declared digits and are returned verbatim either way. + pub truncate_fraction_to_decimals: bool, } #[derive(Debug, Clone, PartialEq)] pub enum Date<'a> { @@ -31,6 +37,9 @@ pub enum JsValue<'a> { Buffer(Cow<'a, [u8]>), Json(&'a str), Date(Date<'a>), + /// A GEOMETRY value (4-byte little-endian SRID, then WKB). mysql2 turns it + /// into point/array objects; like `Json` and `Date` that is host work. + Geometry(&'a [u8]), } fn utf8(b: &[u8]) -> Result<&str> { std::str::from_utf8(b) @@ -52,7 +61,13 @@ fn bigint(value: i128, options: Options) -> JsValue<'static> { } } /// Invoke this default conversion after a host typeCast hook chooses `next()`; -/// raw fields and Column metadata are also available without conversion. +/// raw fields and Column metadata are also available without conversion +/// (`Row::typed` pairs every value with its `ColumnTypeInfo`). +/// +/// The policy is chosen by MySQL column type first. `character_set == 63` +/// (binary) is reported for every non-string column, so it only selects Buffer +/// for the string/BLOB family; TIME stays a string, BIT a Buffer and GEOMETRY a +/// `Geometry` request, matching mysql2. pub fn decode<'a>( info: ColumnTypeInfo, value: RawValue<'a>, @@ -98,7 +113,9 @@ pub fn decode<'a>( JsValue::Json(utf8(bytes)?) } } - 16 | 255 => JsValue::Buffer(bytes.into()), + 11 => JsValue::String(utf8(bytes)?.into()), + 16 => JsValue::Buffer(bytes.into()), + 255 => JsValue::Geometry(bytes), _ if info.character_set == 63 => JsValue::Buffer(bytes.into()), _ => JsValue::String(utf8(bytes)?.into()), }, @@ -129,8 +146,15 @@ pub fn decode<'a>( } else { let mut s = format!("{year:04}-{month:02}-{day:02} {hour:02}:{minute:02}:{second:02}"); - if microsecond != 0 { - s.push_str(&format!(".{microsecond:06}")); + let digits = if options.truncate_fraction_to_decimals { + usize::from(info.decimals.min(6)) + } else { + 6 + }; + if microsecond != 0 && digits > 0 { + let fraction = format!("{microsecond:06}"); + s.push('.'); + s.push_str(&fraction[..digits]); } s }; @@ -169,8 +193,80 @@ mod tests { column_type: t, flags: ColumnFlags::empty(), character_set: 45, + decimals: 0, } } + /// MySQL reports charset 63 for every non-string column; the column type, + /// not the charset, must pick TIME/BIT/GEOMETRY's policy. + #[test] + fn binary_charset_does_not_collapse_time_bit_and_geometry() { + let binary = |t| ColumnTypeInfo { + character_set: 63, + ..info(t) + }; + let decode = |t, bytes| decode(binary(t), RawValue::Bytes(bytes), Options::default()); + assert_eq!( + decode(ColumnType::MYSQL_TYPE_TIME, b"-838:59:59.5"), + Ok(JsValue::String("-838:59:59.5".into())) + ); + assert_eq!( + decode(ColumnType::MYSQL_TYPE_BIT, &[0b101]), + Ok(JsValue::Buffer(Cow::Borrowed(&[0b101]))) + ); + let point = [0, 0, 0, 0, 1, 1, 0, 0, 0]; + assert_eq!( + decode(ColumnType::MYSQL_TYPE_GEOMETRY, &point), + Ok(JsValue::Geometry(&point)) + ); + assert_eq!( + decode(ColumnType::MYSQL_TYPE_VAR_STRING, b"\xff\x00"), + Ok(JsValue::Buffer(Cow::Borrowed(b"\xff\x00"))), + "VARBINARY still follows the binary charset" + ); + } + #[test] + fn datetime_fraction_follows_the_column_decimals_on_request() { + let value = || RawValue::Scalar(Value::Date(2026, 9, 22, 10, 11, 12, 123_456)); + let datetime = |decimals| ColumnTypeInfo { + decimals, + ..info(ColumnType::MYSQL_TYPE_DATETIME) + }; + let strings = Options { + date_strings: true, + ..Options::default() + }; + let truncated = Options { + truncate_fraction_to_decimals: true, + ..strings + }; + assert_eq!( + decode(datetime(3), value(), strings), + Ok(JsValue::String("2026-09-22 10:11:12.123456".into())) + ); + assert_eq!( + decode(datetime(3), value(), truncated), + Ok(JsValue::String("2026-09-22 10:11:12.123".into())) + ); + assert_eq!( + decode(datetime(6), value(), truncated), + Ok(JsValue::String("2026-09-22 10:11:12.123456".into())) + ); + assert_eq!( + decode(datetime(0), value(), truncated), + Ok(JsValue::String("2026-09-22 10:11:12".into())) + ); + assert_eq!( + decode( + ColumnTypeInfo { + decimals: 2, + ..info(ColumnType::MYSQL_TYPE_TIMESTAMP) + }, + value(), + truncated + ), + Ok(JsValue::String("2026-09-22 10:11:12.12".into())) + ); + } #[test] fn node_conversion_options() { let decimal = info(ColumnType::MYSQL_TYPE_NEWDECIMAL); diff --git a/protocols/turnloop-mysql/src/wire.rs b/protocols/turnloop-mysql/src/wire.rs index 611304c..dbe5814 100644 --- a/protocols/turnloop-mysql/src/wire.rs +++ b/protocols/turnloop-mysql/src/wire.rs @@ -51,11 +51,17 @@ impl<'a> Cursor<'a> { } } } +/// Per-column wire metadata that decides how a value is decoded. Rows carry it +/// for every column (`Row::columns`, `Row::typed`), so a host can apply its own +/// policy by MySQL column type without tracking `Column` events. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub struct ColumnTypeInfo { pub column_type: ColumnType, pub flags: ColumnFlags, pub character_set: u16, + /// Declared fractional-second digits for temporal columns (0..=6); 0x1f + /// (31) for columns without a fixed scale. + pub decimals: u8, } #[derive(Debug, Clone, Copy)] pub struct Column<'a> { @@ -102,6 +108,7 @@ impl<'a> Column<'a> { column_type, flags, character_set, + decimals, }, }) } @@ -112,6 +119,14 @@ pub enum RawValue<'a> { Bytes(&'a [u8]), Scalar(Value), } +/// One row packet, decoded lazily and validated as it is iterated. +/// +/// `Row::parse` checks the header (binary marker and NULL bitmap) eagerly; each +/// value is bounds-checked and decoded exactly once as the iterator reaches it, +/// and the final value also rejects trailing packet bytes. The first `Err` ends +/// the iteration (every later `next` is `None`), so nothing past a malformed +/// value is ever yielded. A row error means the server sent a malformed packet: +/// treat it like any other parsing error and `abort` the connection. #[derive(Debug, Clone)] pub struct Row<'a> { cursor: Cursor<'a>, @@ -135,19 +150,27 @@ impl<'a> Row<'a> { } else { &[] }; - let row = Self { + Ok(Self { cursor, columns, bitmap, binary, at: 0, - }; - let mut check = row.clone(); - for v in &mut check { - v?; - } - check.cursor.end()?; - Ok(row) + }) + } + /// Wire metadata of every column in this row, in value order. + pub fn columns(&self) -> &'a [ColumnTypeInfo] { + self.columns + } + /// Pair each remaining value with its column's wire metadata, ready for + /// `types::decode` or a host's own per-type policy. Validation is the same + /// as iterating the row itself. + pub fn typed( + self, + ) -> impl ExactSizeIterator)>> + 'a { + let columns = &self.columns[self.at..]; + self.zip(columns) + .map(|(value, info)| value.map(|value| (*info, value))) } fn value(&mut self, index: usize) -> Result> { if !self.binary { @@ -190,7 +213,17 @@ impl<'a> Iterator for Row<'a> { } let i = self.at; self.at += 1; - Some(self.value(i)) + let value = self.value(i).and_then(|value| { + if self.at == self.columns.len() { + self.cursor.end()?; + } + Ok(value) + }); + if value.is_err() { + // Fuse: never decode past a malformed value. + self.at = self.columns.len(); + } + Some(value) } fn size_hint(&self) -> (usize, Option) { let n = self.columns.len() - self.at; @@ -264,3 +297,111 @@ pub fn error_code(errno: u16) -> Option<&'static str> { _ => return None, }) } +#[cfg(test)] +mod tests { + use super::*; + fn info(column_type: ColumnType) -> ColumnTypeInfo { + ColumnTypeInfo { + column_type, + flags: ColumnFlags::empty(), + character_set: 45, + decimals: 0, + } + } + fn text() -> [ColumnTypeInfo; 2] { + [info(ColumnType::MYSQL_TYPE_VAR_STRING); 2] + } + #[test] + fn a_valid_row_is_decoded_in_one_pass() { + let columns = text(); + let mut row = Row::parse(b"\x02ab\xfb", &columns, false).expect("valid header"); + assert_eq!(row.len(), 2); + assert_eq!(row.next(), Some(Ok(RawValue::Bytes(b"ab")))); + assert_eq!(row.len(), 1); + assert_eq!(row.next(), Some(Ok(RawValue::Null))); + assert_eq!(row.next(), None); + } + /// Values are validated as they are reached; nothing after the first error + /// is decoded or yielded. + #[test] + fn a_malformed_value_ends_the_row() { + let columns = text(); + // Second value claims 9 bytes but only 2 remain. + let mut row = Row::parse(b"\x02ab\x09cd", &columns, false) + .expect("parse no longer decodes values ahead of the host"); + assert_eq!(row.next(), Some(Ok(RawValue::Bytes(b"ab")))); + assert_eq!(row.next(), Some(Err(Error::Protocol("truncated packet")))); + assert_eq!(row.len(), 0); + assert_eq!(row.next(), None); + } + #[test] + fn trailing_bytes_fail_the_last_value() { + let columns = text(); + let mut row = Row::parse(b"\x02ab\x01c!", &columns, false).expect("valid header"); + assert_eq!(row.next(), Some(Ok(RawValue::Bytes(b"ab")))); + assert_eq!( + row.next(), + Some(Err(Error::Protocol("trailing packet bytes"))) + ); + assert_eq!(row.next(), None); + } + #[test] + fn rows_carry_column_type_metadata_alongside_values() { + use mysql_common::proto::MySerialize; + let mut bytes = Vec::new(); + mysql_common::packets::Column::new(ColumnType::MYSQL_TYPE_DATETIME) + .with_decimals(3) + .serialize(&mut bytes); + let column = Column::parse(&bytes).expect("valid column"); + let datetime = column.type_info; + assert_eq!(datetime.column_type, ColumnType::MYSQL_TYPE_DATETIME); + assert_eq!(datetime.decimals, 3); + assert_eq!(column.decimals, 3); + let columns = [ + ColumnTypeInfo { + character_set: 63, + ..info(ColumnType::MYSQL_TYPE_TIME) + }, + ColumnTypeInfo { + character_set: 63, + ..info(ColumnType::MYSQL_TYPE_BIT) + }, + datetime, + ]; + let row = Row::parse(b"\x0801:02:03\x01\x01\xfb", &columns, false).expect("valid"); + assert_eq!(row.columns(), columns); + let typed = row.typed(); + assert_eq!(typed.len(), 3); + let typed: Vec<_> = typed.collect::>().expect("valid row"); + assert_eq!( + typed, + [ + (columns[0], RawValue::Bytes(b"01:02:03")), + (columns[1], RawValue::Bytes(b"\x01")), + (columns[2], RawValue::Null), + ] + ); + } + #[test] + fn binary_rows_validate_the_header_eagerly_and_scalars_lazily() { + let long = [info(ColumnType::MYSQL_TYPE_LONG)]; + assert_eq!( + Row::parse(&[1, 0, 42, 0, 0, 0], &long, true).err(), + Some(Error::Protocol("invalid binary row marker")) + ); + assert_eq!( + Row::parse(&[0], &long, true).err(), + Some(Error::Protocol("truncated packet")) + ); + let mut row = Row::parse(&[0, 0, 42, 0, 0, 0], &long, true).expect("valid row"); + assert_eq!(row.next(), Some(Ok(RawValue::Scalar(Value::Int(42))))); + assert_eq!(row.next(), None); + // A LONG needs four bytes; two are present. + let mut row = Row::parse(&[0, 0, 42, 0], &long, true).expect("valid header"); + assert_eq!( + row.next(), + Some(Err(Error::Protocol("invalid binary scalar"))) + ); + assert_eq!(row.next(), None); + } +} diff --git a/protocols/turnloop-mysql/tests/protocol.rs b/protocols/turnloop-mysql/tests/protocol.rs index 9b3200e..498a73a 100644 --- a/protocols/turnloop-mysql/tests/protocol.rs +++ b/protocols/turnloop-mysql/tests/protocol.rs @@ -468,3 +468,242 @@ fn local_infile_enabled_is_an_explicit_borrowed_request() { }) )); } +/// Prepare `id` with no parameters or columns; the server's reply completes it. +fn prepare_statement(c: &mut Connection, token: Token, id: u32) -> Statement { + c.prepare(token, "DO 1", None) + .expect("fixture operation must succeed"); + flush(c); + let mut prepare = vec![0]; + prepare.extend_from_slice(&id.to_le_bytes()); + prepare.extend_from_slice(&[0, 0, 0, 0, 0, 0, 0]); + c.receive(&frame(1, &prepare)) + .expect("fixture operation must succeed"); + let Some(Event::Prepared { statement, .. }) = + c.next_event().expect("fixture operation must succeed") + else { + panic!("expected Prepared") + }; + assert!(matches!( + c.next_event().expect("fixture operation must succeed"), + Some(Event::Completed { .. }) + )); + statement +} +/// A completion-driven host only calls into the connection when a write or a +/// read completes. COM_STMT_CLOSE and COM_QUIT get no reply, so the write +/// acknowledgement alone must say "call next_event() now"; no timer is used. +#[test] +fn no_reply_commands_complete_from_the_write_acknowledgement() { + let mut c = ready(); + let stmt = prepare_statement(&mut c, 1, 17); + c.close_statement(2, stmt.id) + .expect("fixture operation must succeed"); + let close = c.output().to_vec(); + assert_eq!(close, frame(0, &[0x19, 17, 0, 0, 0])); + assert!(c.next_event().expect("unwritten close").is_none()); + // Partial write completion: nothing to deliver yet. + assert!(!c.consume_output(3).expect("fixture operation must succeed")); + assert!(c.next_event().expect("half-written close").is_none()); + // Final write completion: the only wakeup this command will ever get. + assert!( + c.consume_output(close.len() - 3) + .expect("fixture operation must succeed"), + "acknowledging COM_STMT_CLOSE must ask for next_event()" + ); + assert!(matches!( + c.next_event().expect("fixture operation must succeed"), + Some(Event::Completed { + token: 2, + outcome: Outcome::Success + }) + )); + assert!(c.next_event().expect("drained").is_none()); + assert!(c.can_accept()); + + // A command that does get a reply is not reported early. + c.ping(3).expect("fixture operation must succeed"); + let n = c.output().len(); + assert!(!c.consume_output(n).expect("fixture operation must succeed")); + assert!(c.next_event().expect("awaiting reply").is_none()); + c.receive(&ok(1, 0, 2)) + .expect("fixture operation must succeed"); + while c.next_event().expect("ping reply").is_some() {} + + c.quit().expect("fixture operation must succeed"); + assert_eq!(c.output(), frame(0, &[1])); + assert!(c.next_event().expect("unwritten quit").is_none()); + let n = c.output().len(); + assert!( + c.consume_output(n).expect("fixture operation must succeed"), + "acknowledging COM_QUIT must ask for next_event()" + ); + assert!(matches!( + c.next_event().expect("fixture operation must succeed"), + Some(Event::Closed { + reason: Error::Cancelled + }) + )); + assert!(c.next_event().expect("closed").is_none()); +} +/// The statement stays registered until the COM_STMT_CLOSE bytes are +/// acknowledged, so a failed flush cannot leave the core believing the server +/// dropped a statement it never heard about. +#[test] +fn close_statement_forgets_the_statement_only_after_its_bytes_are_sent() { + let mut c = ready(); + let stmt = prepare_statement(&mut c, 1, 17); + assert_eq!(c.statements(), [stmt]); + c.close_statement(2, stmt.id) + .expect("fixture operation must succeed"); + assert_eq!(c.statements(), [stmt], "close not yet written"); + let n = c.output().len(); + assert!(!c.consume_output(n - 1).expect("partial write")); + assert_eq!(c.statements(), [stmt], "close only partly written"); + // The flush carrying the last byte fails: the host aborts. + c.abort(Error::Transport); + assert_eq!(c.statements(), [stmt], "the server still holds it"); + assert!(matches!( + c.next_event().expect("fixture operation must succeed"), + Some(Event::Completed { + token: 2, + outcome: Outcome::Aborted(Error::Transport) + }) + )); + + let mut c = ready(); + let stmt = prepare_statement(&mut c, 1, 17); + c.close_statement(2, stmt.id) + .expect("fixture operation must succeed"); + let n = c.output().len(); + assert!(c.consume_output(n).expect("full write")); + assert_eq!(c.statements(), []); + assert_eq!( + c.execute(3, stmt.id, &[], None), + Err(Error::State("connection busy or closed")), + "Completed not yet delivered" + ); + assert!(matches!( + c.next_event().expect("fixture operation must succeed"), + Some(Event::Completed { + token: 2, + outcome: Outcome::Success + }) + )); + assert_eq!( + c.execute(3, stmt.id, &[], None), + Err(Error::State("unknown prepared statement")) + ); +} +/// A statement's OK packet and a result set's terminating EOF are different +/// events, so a host never reads an EOF's bytes as affected rows. +#[test] +fn ok_packets_and_result_set_eofs_are_distinct_events() { + let mut c = ready(); + c.query(1, "INSERT INTO t VALUES (1),(2),(3); SELECT 42", None) + .expect("fixture operation must succeed"); + flush(&mut c); + // OK: affected_rows 3, last_insert_id 9, SERVER_MORE_RESULTS_EXISTS | AUTOCOMMIT. + let insert = frame(1, &[0, 3, 9, 10, 0, 0, 0]); + // Legacy EOF: 0xfe, warnings = 3, status = AUTOCOMMIT. + let end = frame(6, &[0xfe, 3, 0, 2, 0]); + c.receive( + &[ + insert, + frame(2, &[1]), + column(3, ColumnType::MYSQL_TYPE_LONG), + eof(4, 2), + frame(5, b"\x0242"), + end, + ] + .concat(), + ) + .expect("fixture operation must succeed"); + let mut seen = Vec::new(); + while let Some(e) = c.next_event().expect("fixture operation must succeed") { + match e { + Event::Ok { token, packet } => seen.push(format!( + "ok {token} affected={} id={:?} more={}", + packet.affected_rows(), + packet.last_insert_id(), + packet + .status_flags() + .contains(StatusFlags::SERVER_MORE_RESULTS_EXISTS) + )), + Event::Eof { + token, + warnings, + status, + } => seen.push(format!( + "eof {token} warnings={warnings} more={}", + status.contains(StatusFlags::SERVER_MORE_RESULTS_EXISTS) + )), + Event::Row { .. } => seen.push("row".into()), + Event::Completed { token, outcome } => { + seen.push(format!("completed {token} {outcome:?}")) + } + _ => {} + } + } + assert_eq!( + seen, + [ + "ok 1 affected=3 id=Some(9) more=true", + "row", + "eof 1 warnings=3 more=false", + "completed 1 Success", + ] + ); + assert_eq!(c.status(), StatusFlags::SERVER_STATUS_AUTOCOMMIT); +} +/// `can_accept` is the admission rule itself: whenever it is false a command is +/// rejected without side effects, and whenever it is true the next one is taken. +#[test] +fn can_accept_reports_the_exact_admission_decision() { + let mut c = Connection::new(Config::default()).expect("fixture operation must succeed"); + assert!(!c.can_accept()); + assert_eq!( + c.ping(1), + Err(Error::State("connection busy or closed")), + "handshake in progress" + ); + let mut c = ready(); + assert!(c.can_accept()); + c.query(1, "SELECT 1", None) + .expect("admitted when can_accept is true"); + let request = c.output().len(); + assert!(!c.can_accept()); + assert!(c.ping(2).is_err()); + assert_eq!( + c.output().len(), + request, + "a rejected command adds no bytes" + ); + c.consume_output(request - 1) + .expect("fixture operation must succeed"); + assert!(!c.can_accept(), "unacknowledged output"); + assert!(c.ping(2).is_err()); + c.consume_output(1).expect("fixture operation must succeed"); + assert!(!c.can_accept(), "command still pending"); + assert!(c.ping(2).is_err()); + c.receive(&ok(1, 0, 2)) + .expect("fixture operation must succeed"); + assert!(matches!( + c.next_event().expect("fixture operation must succeed"), + Some(Event::Ok { token: 1, .. }) + )); + assert!(!c.can_accept(), "Completed not yet delivered"); + assert!(c.ping(2).is_err()); + assert!(matches!( + c.next_event().expect("fixture operation must succeed"), + Some(Event::Completed { + token: 1, + outcome: Outcome::Success + }) + )); + assert!(c.can_accept()); + c.ping(2).expect("admitted when can_accept is true"); + assert_eq!(flush(&mut c), frame(0, &[0x0e])); + c.abort(Error::Transport); + assert!(!c.can_accept()); + assert!(c.quit().is_err()); +} diff --git a/protocols/turnloop-mysql/tests/server.rs b/protocols/turnloop-mysql/tests/server.rs index 26f4b62..616ca63 100644 --- a/protocols/turnloop-mysql/tests/server.rs +++ b/protocols/turnloop-mysql/tests/server.rs @@ -8,6 +8,7 @@ use turnloop_mysql::*; struct Results { rows: Vec>, oks: Vec<(u64, u64, u16)>, + eofs: Vec, errors: Vec<(u16, String, String)>, completed: Vec, statement: Option, @@ -111,6 +112,7 @@ impl Driver { packet.last_insert_id().unwrap_or(0), packet.warnings(), )), + Event::Eof { warnings, .. } => r.eofs.push(warnings), Event::Prepared { statement, .. } => r.statement = Some(statement), Event::Error { error, .. } => r.errors.push(( error.errno, @@ -191,6 +193,8 @@ fn auth_prepared_transactions_compression_and_infile() { let r=d.query("CREATE TEMPORARY TABLE items(id BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY, name TEXT); INSERT INTO items(name) VALUES('one'),('two'); SELECT * FROM items ORDER BY id"); assert_eq!(r.rows.len(), 2); assert!(r.oks.contains(&(2, 1, 0))); + assert_eq!(r.oks.len(), 2, "CREATE and INSERT end in OK packets"); + assert_eq!(r.eofs, [0], "only the SELECT ends in a result-set EOF"); assert_eq!( r.rows[0], vec![Value::Bytes(b"1".to_vec()), Value::Bytes(b"one".to_vec())]