BUG#24595459 - XPLUGIN: SERVER DOES NOT SEEM TO RESPOND TO "0-SIZED" MESSAGES CORRECTLY

Description
===========
When client sends zero length message, X Plugin doesn't respond to that. After next following
messages client receives an error E_X_BAD_MESSAGE and gets disconnected (it depends on next
message header content).

Analyzis
========
The X Protocol message consist from header and payload. The header contains only the
payload lenght (4 bytes). First byte in payload is message type, after which comes the protobuf
binary content. X Plugin always reads 5 bytes, thus sending message with "0-payload" transmits
only the 4 bytes what causes out-of-sync on X Protocol packet layer.

Fix
===

* X Plugin is going to read first 4 bytes, in case of zero lenght it is going to send back
  `Mysqlx.Error` message with error ER_X_BAD_MESSAGE (5000) and text
   "Messages without payload are not supported", in case of valid lenght it is going to read
   whole payload and take the message type from the first byte.

* renamed status variable 'Mysqlx_unknown_message_type' to 'Mysqlx_errors_unknown_message_type'
  which was introduced in BUG#24611754

Reviewed-by: Grzegorz Szwarc <[email protected]>
Reviewed by: Alfredo Kojima <[email protected]>
RB: 14231

Author: lkotula

rapid/plugin/x/mysqlxtest_src/mysqlxtest.cc

@@ -892,9 +892,9 @@ class Command
     m_commands["assert_eq "]      = &Command::cmd_assert_eq;
     m_commands["assert_gt "]      = &Command::cmd_assert_gt;
     m_commands["assert_ge "]      = &Command::cmd_assert_ge;
-    m_commands["query_result"]      = &Command::cmd_query;
-    m_commands["noquery_result"]      = &Command::cmd_noquery;
-    m_commands["wait_for "]      = &Command::cmd_wait_for;
+    m_commands["query_result"]    = &Command::cmd_query;
+    m_commands["noquery_result"]  = &Command::cmd_noquery;
+    m_commands["wait_for "]       = &Command::cmd_wait_for;
   }
 
   bool is_command_syntax(const std::string &cmd) const
@@ -1056,7 +1056,21 @@ class Command
   {
     bool was_set = false;
 
-    cmd_recvresult(context, CMD_ARG_BE_QUIET, ngs::bind(&Command::set_variable, ngs::ref(was_set), args, ngs::placeholders::_1));
+    std::string args_cmd = args;
+    std::vector<std::string> args_array;
+    boost::algorithm::trim(args_cmd);
+
+    boost::split(args_array, args_cmd, boost::is_any_of(" "), boost::token_compress_off);
+
+    args_cmd = CMD_ARG_BE_QUIET;
+
+    if (args_array.size() > 1)
+    {
+      args_cmd += " ";
+      args_cmd += args_array.at(1);
+    }
+
+    cmd_recvresult(context, args_cmd, ngs::bind(&Command::set_variable, ngs::ref(was_set), args_array.at(0), ngs::placeholders::_1));
 
     if (!was_set)
     {
@@ -1560,10 +1574,11 @@ class Command
       return Stop_with_failure;
     }
 
+    context.m_cm->active()->set_closed();
+
     if (context.m_cm->is_default_active())
       return Stop_with_success;
 
-    context.m_cm->active()->set_closed();
     context.m_cm->close_active(false);
 
     return Continue;
@@ -1817,12 +1832,7 @@ class Command
     std::string::size_type p = args.find(' ');
     if (p == std::string::npos)
     {
-      if (variables.find(args) == variables.end())
-      {
-        std::cerr << "Invalid variable " << args << "\n";
-        return Stop_with_failure;
-      }
-      variables.erase(args);
+      variables[args] = "";
     }
     else
     {
@@ -2998,8 +3008,9 @@ class My_command_line_options : public Command_line_options
     std::cout << "  Read and print (if not quiet) one message from the server\n";
     std::cout << "-->recvresult [print-columnsinfo] [" << CMD_ARG_BE_QUIET << "]\n";
     std::cout << "  Read and print one resultset from the server; if print-columnsinfo is present also print short columns status\n";
-    std::cout << "-->recvtovar <varname>\n";
-    std::cout << "  Read and print one resultset from the server and sets the variable from first row\n";
+    std::cout << "-->recvtovar <varname> [COLUMN_NAME]\n";
+    std::cout << "  Read first row and first column (or column with name COLUMN_NAME) of resultset\n";
+    std::cout << "  and set the variable <varname>\n";
     std::cout << "-->recverror <errno>\n";
     std::cout << "  Read a message and ensure that it's an error of the expected type\n";
     std::cout << "-->recvtype <msgtype>\n";

rapid/plugin/x/ngs/include/ngs/protocol_monitor.h

@@ -32,13 +32,14 @@ class Protocol_monitor_interface
 
   virtual void on_notice_warning_send() = 0;
   virtual void on_notice_other_send() = 0;
-  virtual void on_error_send() = 0;
   virtual void on_fatal_error_send() = 0;
   virtual void on_init_error_send() = 0;
   virtual void on_row_send() = 0;
   virtual void on_send(long bytes_transferred) = 0;
   virtual void on_receive(long bytes_transferred) = 0;
-  virtual void on_unknown_msg_type() = 0;
+
+  virtual void on_error_send() = 0;
+  virtual void on_error_unknown_msg_type() = 0;
 };
 
 

rapid/plugin/x/ngs/src/client.cc

@@ -200,7 +200,7 @@ void Client::handle_message(Request &request)
 
     default:
       // invalid message at this time
-      m_protocol_monitor.on_unknown_msg_type();
+      m_protocol_monitor.on_error_unknown_msg_type();
       log_info("%s: Invalid message %i received during client initialization", client_id(), request.get_type());
       m_encoder->send_result(ngs::Fatal(ER_X_BAD_MESSAGE, "Invalid message"));
       m_close_reason = Close_error;
@@ -417,7 +417,7 @@ Request *Client::read_one_message(Error_code &ret_error)
 {
   union
   {
-    char buffer[5];                             // Must be properly aligned
+    char buffer[4];                             // Must be properly aligned
     longlong dummy;
   };
   uint32_t msg_size;
@@ -431,7 +431,7 @@ Request *Client::read_one_message(Error_code &ret_error)
   // untill we get another message to process we mark the connection as idle (for PSF)
   m_connection->mark_idle();
   // read the frame
-  ssize_t nread = m_connection->read(buffer, 5);
+  ssize_t nread = m_connection->read(buffer, 4);
   m_connection->mark_active();
 
   if (nread == 0) // EOF
@@ -460,7 +460,6 @@ Request *Client::read_one_message(Error_code &ret_error)
 #endif
   const uint32_t* pdata = (uint32_t*)(buffer);
   msg_size = *pdata;
-  int8_t type = (int8_t)buffer[4];
 
   if (msg_size > m_server.get_config()->max_message_size)
   {
@@ -471,49 +470,47 @@ Request *Client::read_one_message(Error_code &ret_error)
     return NULL;
   }
 
-  Request_unique_ptr request(ngs::allocate_object<Request>(type));
-
-  if (msg_size > 1)
+  if (0 == msg_size)
   {
-    if (m_msg_buffer_size < msg_size-1)
-    {
-      m_msg_buffer_size = msg_size - 1;
-      ngs::reallocate_array(m_msg_buffer, m_msg_buffer_size, KEY_memory_x_recv_buffer);
-    }
-
-    nread = m_connection->read(&m_msg_buffer[0], msg_size-1);
-    if (nread == 0) // EOF
-    {
-      log_info("%s: peer disconnected while reading message body", client_id());
-      on_network_error(0);
-      return NULL;
-    }
-    if (nread < 0)
-    {
-      int err;
-      std::string strerr;
-      Connection_vio::get_error(err, strerr);
-      log_info("%s: ERROR reading from socket %s (%i)", client_id(), strerr.c_str(), err);
-      on_network_error(err);
-      return NULL;
-    }
-    m_protocol_monitor.on_receive(static_cast<long>(nread));
-
-    request->buffer(m_msg_buffer, msg_size-1);
+    ret_error = Error(ER_X_BAD_MESSAGE, "Messages without payload are not supported");
+    return NULL;
+  }
 
-    ret_error = m_decoder.parse(*request);
-    return request.release();
+  if (m_msg_buffer_size < msg_size)
+  {
+    m_msg_buffer_size = msg_size;
+    ngs::reallocate_array(m_msg_buffer, m_msg_buffer_size, KEY_memory_x_recv_buffer);
   }
-  else if (msg_size == 1)
+
+  nread = m_connection->read(&m_msg_buffer[0], msg_size);
+  if (nread == 0) // EOF
   {
-    ret_error = m_decoder.parse(*request);
-    return request.release();
+    log_info("%s: peer disconnected while reading message body", client_id());
+    on_network_error(0);
+    return NULL;
   }
-  else
+
+  if (nread < 0)
   {
-    ret_error = Error_code(ER_X_BAD_MESSAGE, "Invalid message");
+    int err;
+    std::string strerr;
+    Connection_vio::get_error(err, strerr);
+    log_info("%s: ERROR reading from socket %s (%i)", client_id(), strerr.c_str(), err);
+    on_network_error(err);
     return NULL;
   }
+
+  m_protocol_monitor.on_receive(static_cast<long>(nread));
+
+  int8_t type = (int8_t)m_msg_buffer[0];
+  Request_unique_ptr request(ngs::allocate_object<Request>(type));
+
+  if (msg_size > 1)
+    request->buffer(&m_msg_buffer[1], msg_size - 1);
+
+  ret_error = m_decoder.parse(*request);
+
+  return request.release();
 }
 
 

rapid/plugin/x/ngs/src/client_session.cc

@@ -182,7 +182,7 @@ bool Session::handle_auth_message(ngs::Request &command)
   }
   else
   {
-    m_encoder->get_protocol_monitor().on_unknown_msg_type();
+    m_encoder->get_protocol_monitor().on_error_unknown_msg_type();
     log_info("%s: Unexpected message of type %i received during authentication", m_client.client_id(), type);
     m_encoder->send_init_error(ngs::Fatal(ER_X_BAD_MESSAGE, "Invalid message"));
     stop_auth();

rapid/plugin/x/src/xpl_client.cc

@@ -278,7 +278,7 @@ void Protocol_monitor::on_receive(long bytes_transferred)
 }
 
 
-void Protocol_monitor::on_unknown_msg_type()
+void Protocol_monitor::on_error_unknown_msg_type()
 {
-  update_status<&Common_status_variables::m_unknown_message_type>(m_client->get_session());
+  update_status<&Common_status_variables::m_errors_unknown_message_type>(m_client->get_session());
 }

rapid/plugin/x/src/xpl_client.h

@@ -46,7 +46,7 @@ namespace xpl
     virtual void on_row_send();
     virtual void on_send(long bytes_transferred);
     virtual void on_receive(long bytes_transferred);
-    virtual void on_unknown_msg_type();
+    virtual void on_error_unknown_msg_type();
 
   private:
     Client *m_client;

rapid/plugin/x/src/xpl_common_status_variables.h

@@ -70,7 +70,7 @@ class Common_status_variables
   Variable m_rows_sent;
   Variable m_notice_warning_sent;
   Variable m_notice_other_sent;
-  Variable m_unknown_message_type;
+  Variable m_errors_unknown_message_type;
 
 private:
   Common_status_variables(const Common_status_variables &);

rapid/plugin/x/src/xpl_dispatcher.cc

@@ -193,7 +193,7 @@ ngs::Error_code do_dispatch_command(xpl::Session &session, xpl::Crud_command_han
       return on_expect_close(session, expect, static_cast<const Mysqlx::Expect::Close&>(*command.message()));
   }
 
-  session.proto().get_protocol_monitor().on_unknown_msg_type();
+  session.proto().get_protocol_monitor().on_error_unknown_msg_type();
   return ngs::Error(ER_UNKNOWN_COM_ERROR, "Unexpected message received");
 }
 

rapid/plugin/x/src/xpl_plugin.cc

@@ -245,7 +245,7 @@ static struct st_mysql_show_var xpl_plugin_status[]=
   { MYSQLX_STATUS_VARIABLE_PREFIX("rows_sent"),                      xpl_func_ptr(xpl::Server::common_status_variable<long long, &xpl::Common_status_variables::m_rows_sent>), SHOW_FUNC, SHOW_SCOPE_GLOBAL },
   { MYSQLX_STATUS_VARIABLE_PREFIX("notice_warning_sent"),            xpl_func_ptr(xpl::Server::common_status_variable<long long, &xpl::Common_status_variables::m_notice_warning_sent>), SHOW_FUNC, SHOW_SCOPE_GLOBAL },
   { MYSQLX_STATUS_VARIABLE_PREFIX("notice_other_sent"),              xpl_func_ptr(xpl::Server::common_status_variable<long long, &xpl::Common_status_variables::m_notice_other_sent>), SHOW_FUNC, SHOW_SCOPE_GLOBAL },
-  { MYSQLX_STATUS_VARIABLE_PREFIX("unknown_message_type"),           xpl_func_ptr(xpl::Server::common_status_variable<long long, &xpl::Common_status_variables::m_unknown_message_type>), SHOW_FUNC, SHOW_SCOPE_GLOBAL },
+  { MYSQLX_STATUS_VARIABLE_PREFIX("errors_unknown_message_type"),    xpl_func_ptr(xpl::Server::common_status_variable<long long, &xpl::Common_status_variables::m_errors_unknown_message_type>), SHOW_FUNC, SHOW_SCOPE_GLOBAL },
   { MYSQLX_STATUS_VARIABLE_PREFIX("sessions"),                       xpl_func_ptr(xpl::Server::global_status_variable_server<long long, &xpl::Global_status_variables::m_sessions_count>), SHOW_FUNC, SHOW_SCOPE_GLOBAL },
   { MYSQLX_STATUS_VARIABLE_PREFIX("sessions_closed"),                xpl_func_ptr(xpl::Server::global_status_variable_server<long long, &xpl::Global_status_variables::m_closed_sessions_count>), SHOW_FUNC, SHOW_SCOPE_GLOBAL },
   { MYSQLX_STATUS_VARIABLE_PREFIX("sessions_fatal_error"),           xpl_func_ptr(xpl::Server::global_status_variable_server<long long, &xpl::Global_status_variables::m_sessions_fatal_errors_count>), SHOW_FUNC, SHOW_SCOPE_GLOBAL },