fix(mysqlx): split CURSOR/PREPARE response states for accurate terminal detection

CURSOR_OPEN's terminal set was conflated with CURSOR_FETCH's, and
PREPARE_PREPARE/DEALLOCATE were lumped with PREPARE_EXECUTE which has a wider
terminal set. Adds RESP_WAITING_CURSOR_{OPEN,FETCH,CLOSE} and
RESP_WAITING_PREPARE_{PREPARE,EXECUTE,DEALLOCATE} so each X-Protocol response
shape gets its own per-state contract.

The two old conflated states accepted the union of all terminal frames
that any sub-shape could legitimately emit, so the proxy would advance to
RESP_IDLE on the first such frame regardless of whether the backend's
response was actually shaped right. With the split:
  - PREPARE_PREPARE / PREPARE_DEALLOCATE / CURSOR_CLOSE accept Mysqlx.Ok only
  - PREPARE_EXECUTE accepts Ok / SQL_STMT_EXECUTE_OK / FETCH_DONE / FETCH_SUSPENDED
    (it inherits the response shape of whichever request was prepared)
  - CURSOR_OPEN / CURSOR_FETCH accept FETCH_DONE / FETCH_SUSPENDED

This is purely a tightening of is_terminal_frame; the validation hook
that uses these per-state contracts to actively reject out-of-shape
backend frames is added in the next commit. No behaviour change for
well-behaved backends; existing dispatch tests (66 assertions in
mysqlx_message_dispatch_unit-t, 65 in mysqlx_session_unit-t) continue
to pass unmodified.

Refs: #5694.

plugins/mysqlx/include/mysqlx_session.h

@@ -25,12 +25,30 @@ typedef struct ZSTD_CCtx_s ZSTD_CCtx;
 using MysqlxIdentityLookup =
 	std::function<std::optional<MysqlxResolvedIdentity>(const std::string& username)>;
 
+// Per-message response state for the X-Protocol response sequence the
+// proxy is currently waiting on. The X protocol defines distinct frame
+// allow-sets and terminal markers per request type; this enum splits
+// PREPARE and CURSOR into their per-request sub-shapes so each per-state
+// contract can be expressed cleanly in is_frame_allowed / is_terminal_frame.
+//
+// Background: the previous coarse three-value model (PREPARE / CURSOR
+// each lumped together) over-accepted on PREPARE_PREPARE — which the
+// spec terminates with a bare Mysqlx.Ok — by also allowing the wider
+// SQL_STMT_EXECUTE_OK / FETCH_DONE_* terminators that only PREPARE_EXECUTE
+// can legitimately emit. Likewise CURSOR_OPEN's response always carries
+// ColumnMetaData while CURSOR_FETCH's never does (the metadata is sent
+// once at Open). Splitting the state lets the validation hook reject
+// out-of-shape backend frames precisely.
 enum MysqlxResponseState {
 	RESP_IDLE = 0,
 	RESP_WAITING_STMT_EXECUTE,
 	RESP_WAITING_CRUD,
-	RESP_WAITING_PREPARE,
-	RESP_WAITING_CURSOR,
+	RESP_WAITING_PREPARE_PREPARE,        // Prepare::Prepare — terminator: Ok
+	RESP_WAITING_PREPARE_EXECUTE,        // Prepare::Execute — terminators: Ok / SQL_STMT_EXECUTE_OK / FETCH_DONE / FETCH_SUSPENDED
+	RESP_WAITING_PREPARE_DEALLOCATE,     // Prepare::Deallocate — terminator: Ok
+	RESP_WAITING_CURSOR_OPEN,            // Cursor::Open — terminators: FETCH_DONE / FETCH_SUSPENDED; carries ColumnMetaData
+	RESP_WAITING_CURSOR_FETCH,           // Cursor::Fetch — terminators: FETCH_DONE / FETCH_SUSPENDED; rows-only (metadata was at Open)
+	RESP_WAITING_CURSOR_CLOSE,           // Cursor::Close — terminator: Ok
 	RESP_WAITING_EXPECT,
 	RESP_WAITING_SESS_RESET
 };

plugins/mysqlx/src/mysqlx_session.cpp

@@ -839,16 +839,22 @@ int MysqlxSession::dispatch_client_message(uint8_t msg_type) {
 			forward_to_backend(); return 0;
 		case Mysqlx::ClientMessages_Type_PREPARE_PREPARE:
 			if (backend_conn_) backend_conn_->set_has_prepared_statement(true);
-			response_state_ = RESP_WAITING_PREPARE;
+			response_state_ = RESP_WAITING_PREPARE_PREPARE;
 			forward_to_backend(); return 0;
 		case Mysqlx::ClientMessages_Type_PREPARE_EXECUTE:
+			response_state_ = RESP_WAITING_PREPARE_EXECUTE;
+			forward_to_backend(); return 0;
 		case Mysqlx::ClientMessages_Type_PREPARE_DEALLOCATE:
-			response_state_ = RESP_WAITING_PREPARE;
+			response_state_ = RESP_WAITING_PREPARE_DEALLOCATE;
 			forward_to_backend(); return 0;
 		case Mysqlx::ClientMessages_Type_CURSOR_OPEN:
+			response_state_ = RESP_WAITING_CURSOR_OPEN;
+			forward_to_backend(); return 0;
 		case Mysqlx::ClientMessages_Type_CURSOR_FETCH:
+			response_state_ = RESP_WAITING_CURSOR_FETCH;
+			forward_to_backend(); return 0;
 		case Mysqlx::ClientMessages_Type_CURSOR_CLOSE:
-			response_state_ = RESP_WAITING_CURSOR;
+			response_state_ = RESP_WAITING_CURSOR_CLOSE;
 			forward_to_backend(); return 0;
 		case Mysqlx::ClientMessages_Type_EXPECT_OPEN:
 		case Mysqlx::ClientMessages_Type_EXPECT_CLOSE:
@@ -950,11 +956,31 @@ bool MysqlxSession::is_terminal_frame(uint8_t msg_type) const {
 			return msg_type == Mysqlx::ServerMessages_Type_OK ||
 			       msg_type == Mysqlx::ServerMessages_Type_RESULTSET_FETCH_DONE ||
 			       msg_type == Mysqlx::ServerMessages_Type_RESULTSET_FETCH_SUSPENDED;
-		case RESP_WAITING_PREPARE:
+		case RESP_WAITING_PREPARE_PREPARE:
+			// Prepare::Prepare returns only Mysqlx.Ok on success.
+			return msg_type == Mysqlx::ServerMessages_Type_OK;
+		case RESP_WAITING_PREPARE_EXECUTE:
+			// Prepare::Execute behaves like the underlying request — for
+			// statement preparations the terminator is SQL_STMT_EXECUTE_OK,
+			// for CRUD it's Ok, for cursor-bound preparations it's
+			// FETCH_DONE / FETCH_SUSPENDED. Accept all four; the proxy
+			// can't tell at this layer which kind was prepared.
+			return msg_type == Mysqlx::ServerMessages_Type_OK ||
+			       msg_type == Mysqlx::ServerMessages_Type_SQL_STMT_EXECUTE_OK ||
+			       msg_type == Mysqlx::ServerMessages_Type_RESULTSET_FETCH_DONE ||
+			       msg_type == Mysqlx::ServerMessages_Type_RESULTSET_FETCH_SUSPENDED;
+		case RESP_WAITING_PREPARE_DEALLOCATE:
+			// Prepare::Deallocate returns only Mysqlx.Ok.
 			return msg_type == Mysqlx::ServerMessages_Type_OK;
-		case RESP_WAITING_CURSOR:
+		case RESP_WAITING_CURSOR_OPEN:
 			return msg_type == Mysqlx::ServerMessages_Type_RESULTSET_FETCH_DONE ||
 			       msg_type == Mysqlx::ServerMessages_Type_RESULTSET_FETCH_SUSPENDED;
+		case RESP_WAITING_CURSOR_FETCH:
+			return msg_type == Mysqlx::ServerMessages_Type_RESULTSET_FETCH_DONE ||
+			       msg_type == Mysqlx::ServerMessages_Type_RESULTSET_FETCH_SUSPENDED;
+		case RESP_WAITING_CURSOR_CLOSE:
+			// Cursor::Close returns only Mysqlx.Ok.
+			return msg_type == Mysqlx::ServerMessages_Type_OK;
 		case RESP_WAITING_EXPECT:
 			return msg_type == Mysqlx::ServerMessages_Type_OK;
 		case RESP_WAITING_SESS_RESET: