From c8a7b7cc85aecf6c1f3efb3015537e609e37a036 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ralph=20K=C3=BCpper?= Date: Tue, 22 Sep 2026 19:30:13 +0200 Subject: [PATCH 1/6] Expose the MySQL admission rule as can_accept() accept() admitted a command only when the session was Ready, no command was pending and output() was empty, but that rule was private. A host running its own command queue (Perry's P7 lane) had to rebuild it by hand, including the output().is_empty() flush coupling, and keep the copy in sync. can_accept() is now the public rule and accept() is defined in terms of it, so the two cannot drift: every command method (query, prepare, execute, the simple commands, change_user, quit) still goes through accept(). The new protocol test walks a connection through handshake, submission, partial and full acknowledgement, pending reply, undelivered Completed, idle and aborted, and checks at each point that can_accept() matches whether a command is actually taken (and that a rejected one appends no bytes). Closes #65 --- protocols/turnloop-mysql/README.md | 4 +- protocols/turnloop-mysql/src/lib.rs | 10 ++++- protocols/turnloop-mysql/tests/protocol.rs | 52 ++++++++++++++++++++++ 3 files changed, 64 insertions(+), 2 deletions(-) diff --git a/protocols/turnloop-mysql/README.md b/protocols/turnloop-mysql/README.md index 3a4491f..e769515 100644 --- a/protocols/turnloop-mysql/README.md +++ b/protocols/turnloop-mysql/README.md @@ -61,7 +61,9 @@ the byte stream after it. Each accepted command yields exactly one Completed; server Error and result Ok 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 diff --git a/protocols/turnloop-mysql/src/lib.rs b/protocols/turnloop-mysql/src/lib.rs index 4e3d869..e9093e7 100644 --- a/protocols/turnloop-mysql/src/lib.rs +++ b/protocols/turnloop-mysql/src/lib.rs @@ -403,8 +403,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(()) diff --git a/protocols/turnloop-mysql/tests/protocol.rs b/protocols/turnloop-mysql/tests/protocol.rs index 9b3200e..e770af4 100644 --- a/protocols/turnloop-mysql/tests/protocol.rs +++ b/protocols/turnloop-mysql/tests/protocol.rs @@ -468,3 +468,55 @@ fn local_infile_enabled_is_an_explicit_borrowed_request() { }) )); } +/// `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()); +} From 4fbb1ed4cd57f60c02f664999411da836e4a58e8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ralph=20K=C3=BCpper?= Date: Tue, 22 Sep 2026 19:31:20 +0200 Subject: [PATCH 2/6] Let the output acknowledgement complete MySQL's no-reply commands COM_STMT_CLOSE and COM_QUIT get no server reply. Their NoResponse -> Completed and Closing -> Closed transitions were only taken on a later next_event() call made after output() had been acknowledged, and nothing ever prompted that call: a completion-driven host has no read to react to, so the queue wedged until something else woke the connection. Perry's P7 lane worked around it with a 0 ms turnloop deadline purely to manufacture that wakeup. Of the two fixes the issue offers, consume_output() now returns bool rather than adding a separate poll_pending(). The acknowledgement is the one call a completion-driven host is guaranteed to make at that moment, so it is where the signal belongs; a second query method would still need the host to remember to ask it after every write. The transition itself also moved into consume_output(): acknowledging the last byte of a COM_STMT_CLOSE queues its Completed immediately, and the return value is true whenever an event (Completed or the terminal Closed) is deliverable without further input. The turnloop-io adapter drops the flag because its driver already polls after every drain. The new test drives both commands as a completion-driven host would, with no handle_timeout call: a partial write reports false and delivers nothing, the final write reports true and next_event() yields Completed / Closed, and a ping (which does get a reply) is not reported early. Breaking: Connection::consume_output returns Result instead of Result<()>. Closes #63 --- protocols/turnloop-mysql/README.md | 5 +- protocols/turnloop-mysql/src/asynchronous.rs | 4 +- protocols/turnloop-mysql/src/lib.rs | 17 +++-- protocols/turnloop-mysql/tests/protocol.rs | 77 ++++++++++++++++++++ 4 files changed, 96 insertions(+), 7 deletions(-) diff --git a/protocols/turnloop-mysql/README.md b/protocols/turnloop-mysql/README.md index e769515..4998fc8 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. 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 e9093e7..ec5e373 100644 --- a/protocols/turnloop-mysql/src/lib.rs +++ b/protocols/turnloop-mysql/src/lib.rs @@ -270,7 +270,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 +285,11 @@ impl Connection { if self.output_at == self.output.len() { self.output.clear(); self.output_at = 0; + if self.state == State::NoResponse { + 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!( @@ -650,9 +660,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 diff --git a/protocols/turnloop-mysql/tests/protocol.rs b/protocols/turnloop-mysql/tests/protocol.rs index e770af4..710358b 100644 --- a/protocols/turnloop-mysql/tests/protocol.rs +++ b/protocols/turnloop-mysql/tests/protocol.rs @@ -468,6 +468,83 @@ 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()); +} /// `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] From 2046be673e4d547469d95138544ae01051f730c1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ralph=20K=C3=BCpper?= Date: Tue, 22 Sep 2026 19:31:57 +0200 Subject: [PATCH 3/6] Forget a closed MySQL statement only once its close bytes are sent close_statement() removed the statement from Connection::statements as soon as COM_STMT_CLOSE was queued, before any byte was acknowledged. If the flush carrying it failed, the core believed the statement was gone while the server still held it open. The pending command now records CommandKind::CloseStatement(id), and the statement is removed in consume_output() at the same point the no-reply command completes: when the last byte of its output is acknowledged. A partial write, or an abort after a failed flush, leaves it registered. A new statements() accessor exposes the registry so a host can see which statements the server still holds. The test checks the registry before the write, after a partial write and after an abort (still listed; Completed is Aborted), and after a full write (gone, and execute then reports "unknown prepared statement"). With the old eager retain() it fails at the first "close not yet written" assertion. Closes #67 --- protocols/turnloop-mysql/src/lib.rs | 27 ++++++++++-- protocols/turnloop-mysql/tests/protocol.rs | 49 ++++++++++++++++++++++ 2 files changed, 73 insertions(+), 3 deletions(-) diff --git a/protocols/turnloop-mysql/src/lib.rs b/protocols/turnloop-mysql/src/lib.rs index ec5e373..fb4fa78 100644 --- a/protocols/turnloop-mysql/src/lib.rs +++ b/protocols/turnloop-mysql/src/lib.rs @@ -193,6 +193,8 @@ enum CommandKind { Execute, Reset, ChangeUser, + /// COM_STMT_CLOSE for this statement ID; forgotten once its bytes are sent. + CloseStatement(u32), Other, } struct Pending { @@ -286,6 +288,16 @@ impl Connection { 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); } } @@ -555,9 +567,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, diff --git a/protocols/turnloop-mysql/tests/protocol.rs b/protocols/turnloop-mysql/tests/protocol.rs index 710358b..86c8ab3 100644 --- a/protocols/turnloop-mysql/tests/protocol.rs +++ b/protocols/turnloop-mysql/tests/protocol.rs @@ -545,6 +545,55 @@ fn no_reply_commands_complete_from_the_write_acknowledgement() { )); 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")) + ); +} /// `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] From 7b1df094ba3cd098e8ed0f200c0c3312c4129c8d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ralph=20K=C3=BCpper?= Date: Tue, 22 Sep 2026 19:32:56 +0200 Subject: [PATCH 4/6] Split MySQL's result-set EOF out of Event::Ok Event::Ok fired both for a genuine OK packet and for the legacy EOF that ends a result set's rows. The EOF's OkPacket view reports affected_rows and last_insert_id that are not row counts at all, and nothing in the event said which case produced it, so every host had to shadow the protocol state to interpret the fields. Event::Ok now means only a real OK packet. The row terminator becomes Event::Eof { token, warnings, status }: it exposes exactly what a legacy EOF carries (warning count and status flags, including SERVER_MORE_RESULTS_EXISTS) and deliberately no OkPacket, so there is no meaningless affected_rows to misread. The connection negotiates legacy EOF only (CLIENT_DEPRECATE_EOF is never requested), so this is the only terminator shape. The new protocol test feeds "INSERT ...; SELECT 42" and asserts the exact event sequence: Ok(affected 3, id 9, more results), Row, Eof(warnings 3, no more results), Completed. The ignored real-server suite now also counts Eof separately and asserts that CREATE+INSERT+SELECT yields two OKs and one EOF. Breaking: result-set terminators are Event::Eof, no longer Event::Ok. Closes #64 --- protocols/turnloop-mysql/README.md | 6 ++- protocols/turnloop-mysql/src/lib.rs | 21 ++++++-- protocols/turnloop-mysql/tests/protocol.rs | 61 ++++++++++++++++++++++ protocols/turnloop-mysql/tests/server.rs | 4 ++ 4 files changed, 86 insertions(+), 6 deletions(-) diff --git a/protocols/turnloop-mysql/README.md b/protocols/turnloop-mysql/README.md index 4998fc8..d616f2c 100644 --- a/protocols/turnloop-mysql/README.md +++ b/protocols/turnloop-mysql/README.md @@ -61,7 +61,7 @@ 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; @@ -70,7 +70,9 @@ 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, diff --git a/protocols/turnloop-mysql/src/lib.rs b/protocols/turnloop-mysql/src/lib.rs index fb4fa78..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, @@ -938,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) @@ -948,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/tests/protocol.rs b/protocols/turnloop-mysql/tests/protocol.rs index 86c8ab3..498a73a 100644 --- a/protocols/turnloop-mysql/tests/protocol.rs +++ b/protocols/turnloop-mysql/tests/protocol.rs @@ -594,6 +594,67 @@ fn close_statement_forgets_the_statement_only_after_its_bytes_are_sent() { 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] 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())] From ae50becd08bb7efb4dc36e5769b8375fd3b8c609 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ralph=20K=C3=BCpper?= Date: Tue, 22 Sep 2026 19:34:02 +0200 Subject: [PATCH 5/6] Validate MySQL rows as they are iterated instead of decoding them twice Row::parse validated a row by cloning it and running the clone to the end, then handed the original to the host, which decoded every value again. Every row, text or binary, paid its full decode cost twice. Row::parse now checks only the header eagerly (binary marker and the NULL bitmap length, which is what makes the later bitmap indexing in-bounds) and the iterator validates as it goes: each value is bounds-checked and decoded exactly once when reached, the last value additionally rejects trailing packet bytes, and the first error fuses the iterator (len() drops to 0 and next() returns None), so nothing past a malformed value is ever decoded or yielded. No value is read outside the packet and nothing panics, as before. The observable difference is where a malformed row surfaces: in the host's iteration rather than as next_event()'s Err; the README now says to treat it like any other parsing error and abort. The new unit tests cover a valid row, a truncated value (previously rejected by Row::parse; now the first value is yielded, the second errors and the iterator ends), trailing bytes on the last value, and binary rows with an eagerly rejected marker/bitmap and a lazily rejected short scalar. Closes #66 --- protocols/turnloop-mysql/README.md | 3 + protocols/turnloop-mysql/src/wire.rs | 100 ++++++++++++++++++++++++--- 2 files changed, 94 insertions(+), 9 deletions(-) diff --git a/protocols/turnloop-mysql/README.md b/protocols/turnloop-mysql/README.md index d616f2c..d8a3017 100644 --- a/protocols/turnloop-mysql/README.md +++ b/protocols/turnloop-mysql/README.md @@ -101,6 +101,9 @@ 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. diff --git a/protocols/turnloop-mysql/src/wire.rs b/protocols/turnloop-mysql/src/wire.rs index 611304c..c4cfdd3 100644 --- a/protocols/turnloop-mysql/src/wire.rs +++ b/protocols/turnloop-mysql/src/wire.rs @@ -112,6 +112,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 +143,13 @@ 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) + }) } fn value(&mut self, index: usize) -> Result> { if !self.binary { @@ -190,7 +192,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 +276,73 @@ 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, + } + } + 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 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); + } +} From 973bc7dac38f9ad978e2f28515ab52a9a270d20e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ralph=20K=C3=BCpper?= Date: Tue, 22 Sep 2026 19:35:26 +0200 Subject: [PATCH 6/6] Let types::decode express mysql2's per-column-type wire policy types::decode picked Buffer for any column reporting character_set 63, but MySQL reports the binary charset for every non-string column. A text-protocol TIME therefore decoded as a Buffer instead of mysql2's string, and TIME, BIT and GEOMETRY all fell into the same arm with nothing telling them apart. There was also no way to reproduce mysql2's DATETIME fraction handling: its binary parser cuts the fractional seconds to the column's declared decimals (DATETIME(3) -> ".123"), while decode always printed six digits. Perry's P7 lane ended up not using decode at all. The issue allows either a type-aware decode path plus a truncation option, or at minimum raw metadata beside each value. This does both, since the metadata is cheap and lets a host diverge per type without re-decoding: - decode chooses by column type before charset: TIME is a string, BIT a Buffer (as in mysql2) and GEOMETRY a new JsValue::Geometry request carrying the SRID+WKB bytes, like Json and Date, for the host to build mysql2's objects. Charset 63 now only selects Buffer for the string/BLOB family. - Options::truncate_fraction_to_decimals cuts formatted DATETIME/TIMESTAMP fractions to the column's decimals (0 drops the fraction); off by default, so existing output is unchanged. - ColumnTypeInfo gains `decimals` (filled from the column definition), and Row::columns() / Row::typed() pair every value with its ColumnTypeInfo. Tests: TIME/BIT/GEOMETRY/VARBINARY with charset 63 decode to String / Buffer / Geometry / Buffer (TIME was Buffer before); DATETIME(3), (6), (0) and TIMESTAMP(2) fractions with and without truncation; and a row carrying the parsed DATETIME(3) decimals through Row::typed. Breaking: ColumnTypeInfo has a new public `decimals` field, Options a new field, and JsValue a new Geometry variant. Closes #68 --- protocols/turnloop-mysql/README.md | 15 ++-- protocols/turnloop-mysql/src/types.rs | 104 +++++++++++++++++++++++++- protocols/turnloop-mysql/src/wire.rs | 59 +++++++++++++++ 3 files changed, 169 insertions(+), 9 deletions(-) diff --git a/protocols/turnloop-mysql/README.md b/protocols/turnloop-mysql/README.md index d8a3017..8b03647 100644 --- a/protocols/turnloop-mysql/README.md +++ b/protocols/turnloop-mysql/README.md @@ -105,17 +105,22 @@ 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 | @@ -124,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/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 c4cfdd3..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, }, }) } @@ -151,6 +158,20 @@ impl<'a> Row<'a> { at: 0, }) } + /// 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 { if self.cursor.0.first() == Some(&0xfb) { @@ -284,6 +305,7 @@ mod tests { column_type, flags: ColumnFlags::empty(), character_set: 45, + decimals: 0, } } fn text() -> [ColumnTypeInfo; 2] { @@ -324,6 +346,43 @@ mod tests { 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!(