test(mysql): clean-room integration tests + prepared-stmt error fix + Controller cleanup

- Add clean-room integration tests (transactions, prepared statements, pooled
  connection concurrency, connection-type) run against a self-spawned mysqld.
- Fix: a failed COM_STMT_PREPARE now returns the ERR packet to the caller and keeps
  the connection alive, instead of closing the socket.
- Warn when a prepared statement runs on a 'short' connection (re-prepares on every
  execute; prefer 'pooled').
- Replace Controller's mysql-specific _mysql_stmt with a generic opaque per-RPC slot
  so no protocol type leaks onto the shared Controller.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: rajvarun77

src/brpc/controller.cpp

@@ -298,7 +298,7 @@ void Controller::ResetPods() {
     _response_streams.clear();
     _remote_stream_settings = NULL;
     _bind_sock_action = BIND_SOCK_NONE;
-    _mysql_stmt = NULL;
+    _session_data = NULL;
     _auth_flags = 0;
     _rpc_received_us = 0;
 }
@@ -829,7 +829,8 @@ void Controller::Call::OnComplete(
         if (sending_sock != NULL && (error_code == 0 || responded)) {
             if (bind_sock_action == BIND_SOCK_RESERVE) {
                 // Reserve this socket on the controller for a following RPC
-                // (mysql transaction / prepared statement connection affinity).
+                // (mysql transaction connection affinity; prepared statements
+                // do NOT reserve -- they use a per-socket stmt_id map + re-prepare).
                 c->_bind_sock.reset(sending_sock.release());
             } else if (bind_sock_action == BIND_SOCK_USE) {
                 // Socket is owned by the binder; do not return it to the pool.
@@ -1110,7 +1111,7 @@ void Controller::IssueRPC(int64_t start_realtime_us) {
     SocketUniquePtr tmp_sock;
     if ((_connection_type & CONNECTION_TYPE_POOLED_AND_SHORT) &&
         _bind_sock_action == BIND_SOCK_USE) {
-        // Reuse the socket reserved by a previous RPC (mysql tx/stmt affinity).
+        // Reuse the socket reserved by a previous RPC (mysql transaction affinity).
         tmp_sock.reset(_bind_sock.release());
         if (!tmp_sock || (!is_health_check_call() && !tmp_sock->IsAvailable())) {
             // NOTE: tmp_sock may be NULL here, so guard the id() deref.

src/brpc/controller.h

@@ -108,15 +108,14 @@ enum StopStyle {
 const int32_t UNSET_MAGIC_NUM = -123456789;
 
 // If a controller wants to reserve the sending socket after the RPC (e.g. mysql
-// transactions/prepared statements that need connection affinity), set
-// BIND_SOCK_RESERVE; later RPCs reuse it via BIND_SOCK_USE.
+// transactions that need connection affinity), set BIND_SOCK_RESERVE; later RPCs
+// reuse it via BIND_SOCK_USE. (Prepared statements do NOT reserve -- they use a
+// per-socket stmt_id map + re-prepare instead of pinning a connection.)
 enum BindSockAction {
     BIND_SOCK_RESERVE,
     BIND_SOCK_USE,
     BIND_SOCK_NONE,
 };
-// mysql prepared statement, defined in mysql.h
-class MysqlStatementStub;
 
 typedef butil::FlatMap<std::string, std::string> UserFieldsMap;
 
@@ -927,13 +926,15 @@ friend void policy::ProcessThriftRequest(InputMessageBase*);
     // Defined at both sides
     StreamSettings *_remote_stream_settings;
 
-    // Whether/how to reserve the sending socket after the RPC (mysql tx/stmt).
+    // Whether/how to reserve the sending socket after the RPC (mysql transactions).
     BindSockAction _bind_sock_action;
     // The socket reserved by a previous RPC and reused when _bind_sock_action
     // is BIND_SOCK_USE.
     SocketUniquePtr _bind_sock;
-    // mysql prepared statement bound to this RPC, owned elsewhere.
-    MysqlStatementStub* _mysql_stmt;
+    // Opaque per-RPC slot a protocol codec may use to carry typed state from
+    // serialize_request to pack_request/parse (e.g. the mysql prepared-statement
+    // stub). Not owned by Controller.
+    void* _session_data;
 
     // Thrift method name, only used when thrift protocol enabled
     std::string _thrift_method_name;

src/brpc/details/controller_private_accessor.h

@@ -134,7 +134,7 @@ class ControllerPrivateAccessor {
 
     void clear_auth_flags() { _cntl->_auth_flags = 0; }
 
-    // Set how the sending socket is reserved after the RPC (mysql tx/stmt).
+    // Set how the sending socket is reserved after the RPC (mysql transactions).
     void set_bind_sock_action(BindSockAction action) { _cntl->_bind_sock_action = action; }
     // Transfer ownership of the reserved socket to `ptr`.
     void get_bind_sock(SocketUniquePtr* ptr) {
@@ -147,10 +147,8 @@ class ControllerPrivateAccessor {
         _cntl->_bind_sock_action = BIND_SOCK_USE;
         Socket::Address(sock_id, &_cntl->_bind_sock);
     }
-    // Set the mysql prepared statement bound to this RPC.
-    void set_mysql_stmt(MysqlStatementStub* stmt) { _cntl->_mysql_stmt = stmt; }
-    // Get the mysql prepared statement bound to this RPC.
-    MysqlStatementStub* mysql_stmt() { return _cntl->_mysql_stmt; }
+    void set_session_data(void* d) { _cntl->_session_data = d; }
+    void* session_data() const { return _cntl->_session_data; }
 
     std::string& protocol_param() { return _cntl->protocol_param(); }
     const std::string& protocol_param() const { return _cntl->protocol_param(); }

src/brpc/mysql_statement.cpp

@@ -69,6 +69,12 @@ void MysqlStatement::Init(const Channel& channel) {
     _connection_type = ConnectionType(opts.connection_type);
     if (_connection_type != CONNECTION_TYPE_SHORT) {
         _id_map.Modify(my_init_kv);
+    } else {
+        LOG_EVERY_SECOND(WARNING)
+            << "Prepared statement on a 'short' connection re-prepares on every "
+               "execute (a new TCP connection per request cannot cache the "
+               "server stmt_id); use connection_type='pooled' for prepared "
+               "statements.";
     }
 }
 

src/brpc/policy/mysql_protocol.cpp

@@ -107,7 +107,7 @@ bool PackRequest(butil::IOBuf* buf,
             LOG(ERROR) << "[MYSQL PACK] get sending socket with NULL";
             return false;
         }
-        auto stub = accessor.mysql_stmt();
+        auto stub = static_cast<MysqlStatementStub*>(accessor.session_data());
         if (stub == NULL) {
             LOG(ERROR) << "[MYSQL PACK] get prepare statement with NULL";
             return false;
@@ -284,7 +284,7 @@ ParseError HandlePrepareStatement(const InputResponse* msg,
     ParseError parseCode = PARSE_OK;
     butil::IOBuf buf;
     butil::Status st;
-    auto stub = ControllerPrivateAccessor(cntl).mysql_stmt();
+    auto stub = static_cast<MysqlStatementStub*>(ControllerPrivateAccessor(cntl).session_data());
     auto stmt = stub->stmt();
     if (stmt == NULL || stmt->param_count() != ok.param_count()) {
         LOG(ERROR) << "[MYSQL PACK] stmt can't be NULL";
@@ -359,6 +359,16 @@ ParseResult ParseMysqlMessage(butil::IOBuf* source,
         return MakeParseError(PARSE_ERROR_NOT_ENOUGH_DATA);
     }
     if (stmt_type == MYSQL_NEED_PREPARE) {
+        // A failed PREPARE (e.g. ER_PARSE_ERROR 1064) comes back as a normal
+        // ERR packet. Deliver it to the caller like any other error response
+        // and keep the connection open -- matching the command path and other
+        // protocols (redis, baidu_std). Only a successful prepare proceeds to
+        // pack and send the COM_STMT_EXECUTE.
+        if (!msg->response.reply(0).is_prepare_ok()) {
+            msg->id_wait = pi.id_wait;
+            socket->release_parsing_context();
+            return MakeMessage(msg);
+        }
         // store stmt_id, make execute header.
         ParseError err = HandlePrepareStatement(msg, socket, &pi);
         if (err != PARSE_OK) {
@@ -426,10 +436,11 @@ void SerializeMysqlRequest(butil::IOBuf* buf,
     if (!rr->SerializeTo(buf)) {
         return cntl->SetFailed(EREQUEST, "Fail to serialize MysqlRequest");
     }
-    // mysql protocol don't use pipelined count to verify the end of a response, so pipelined count
-    // is meanless, but we can use it help us to distinguish mysql reply type. In mysql protocol, we
-    // can't distinguish OK and PreparedOk, so we set pipelined count to 2 to let parse function to
-    // parse PreparedOk reply
+    // mysql doesn't use pipelined_count to verify the end of a response; instead we
+    // reuse it as a MysqlStmtType tag so the parse function knows which reply shape
+    // to expect (OK and PrepareOk are otherwise indistinguishable). Default to
+    // MYSQL_NORMAL_STATEMENT (1); it is upgraded to MYSQL_PREPARED_STATEMENT (2)
+    // below when the request carries a prepared statement.
     ControllerPrivateAccessor accessor(cntl);
     accessor.set_pipelined_count(MYSQL_NORMAL_STATEMENT);
 
@@ -439,7 +450,7 @@ void SerializeMysqlRequest(butil::IOBuf* buf,
     }
     auto st = rr->get_stmt();
     if (st != NULL) {
-        accessor.set_mysql_stmt(st);
+        accessor.set_session_data(rr->get_stmt());
         accessor.set_pipelined_count(MYSQL_PREPARED_STATEMENT);
     }
     if (FLAGS_mysql_verbose) {