Merge pull request #5203 from sysown/v3.0_FastForwardGracefulClose

renecannao · web-flow · commit 29a6b8576a6c · 2025-11-20T12:37:36.000+07:00
Implement Fast Forward Grace Close Feature to Prevent Data Loss
diff --git a/include/MySQL_Data_Stream.h b/include/MySQL_Data_Stream.h
@@ -176,6 +176,9 @@ class MySQL_Data_Stream
 	char kill_type;
 
 	bool encrypted;
+	// defer_close_due_to_fast_forward: Flag to prevent immediate closure of data stream
+	// during fast forward grace close, allowing buffers to drain.
+	bool defer_close_due_to_fast_forward;
 	bool net_failure;
 
 	uint8_t pkt_sid;
diff --git a/include/MySQL_Session.h b/include/MySQL_Session.h
@@ -403,6 +403,10 @@ class MySQL_Session: public Base_Session<MySQL_Session, MySQL_Data_Stream, MySQL
 
 	ProxySQL_Node_Address * proxysql_node_address; // this is used ONLY for Admin, and only if the other party is another proxysql instance part of a cluster
 	bool use_ldap_auth;
+	// Fast forward grace close flags: track backend closure during fast forward mode
+	// to allow pending client data to drain before closing the session.
+	bool backend_closed_in_fast_forward;
+	unsigned long long fast_forward_grace_start_time;
 
 	// this variable is relevant only if status == SETTING_VARIABLE
 	enum mysql_variable_name changing_variable_idx;
diff --git a/include/MySQL_Thread.h b/include/MySQL_Thread.h
@@ -316,9 +316,11 @@ struct p_th_gauge {
 		mysql_monitor_replication_lag_timeout,
 		mysql_monitor_history,
 		__size
+
 	};
 };
 
+
 struct th_metrics_map_idx {
 	enum index {
 		counters = 0,
@@ -428,6 +430,7 @@ class MySQL_Threads_Handler
 		char * monitor_replication_lag_use_percona_heartbeat;
 		int ping_interval_server_msec;
 		int ping_timeout_server;
+		int fast_forward_grace_close_ms;
 		int shun_on_failures;
 		int shun_recovery_time_sec;
 		int unshun_algorithm;
diff --git a/include/proxysql_structs.h b/include/proxysql_structs.h
@@ -1226,6 +1226,7 @@ __thread int mysql_thread___throttle_connections_per_sec_to_hostgroup;
 __thread int mysql_thread___max_transaction_idle_time;
 __thread int mysql_thread___max_transaction_time;
 __thread int mysql_thread___threshold_query_length;
+__thread int mysql_thread___fast_forward_grace_close_ms;
 __thread int mysql_thread___threshold_resultset_size;
 __thread int mysql_thread___wait_timeout;
 __thread int mysql_thread___throttle_max_bytes_per_second_to_client;
@@ -1529,6 +1530,7 @@ extern __thread int mysql_thread___throttle_connections_per_sec_to_hostgroup;
 extern __thread int mysql_thread___max_transaction_idle_time;
 extern __thread int mysql_thread___max_transaction_time;
 extern __thread int mysql_thread___threshold_query_length;
+extern __thread int mysql_thread___fast_forward_grace_close_ms;
 extern __thread int mysql_thread___threshold_resultset_size;
 extern __thread int mysql_thread___wait_timeout;
 extern __thread int mysql_thread___throttle_max_bytes_per_second_to_client;
diff --git a/lib/MySQL_Session.cpp b/lib/MySQL_Session.cpp
@@ -684,6 +684,8 @@ MySQL_Session::MySQL_Session() {
 	last_HG_affected_rows = -1; // #1421 : advanced support for LAST_INSERT_ID()
 	proxysql_node_address = NULL;
 	use_ldap_auth = false;
+	backend_closed_in_fast_forward = false;
+	fast_forward_grace_start_time = 0;
 }
 
 /**
@@ -715,6 +717,8 @@ void MySQL_Session::reset() {
 	mybe=NULL;
 
 	with_gtid = false;
+	backend_closed_in_fast_forward = false;
+	fast_forward_grace_start_time = 0;
 
 	//gtid_trxid = 0;
 	gtid_hid = -1;
@@ -3778,6 +3782,38 @@ int MySQL_Session::GPFC_Statuses2(bool& wrong_pass, PtrSize_t& pkt) {
 			break;
 		case FAST_FORWARD:
 			mybe->server_myds->PSarrayOUT->add(pkt.ptr, pkt.size);
+			/*
+			 * Fast Forward Grace Close Logic:
+			 * In fast forward mode, ProxySQL forwards packets without buffering them.
+			 * If the backend connection closes unexpectedly while client data is still
+			 * being sent, we risk data loss. To mitigate this, we implement a grace
+			 * period where the session remains open until all pending client output
+			 * buffers are drained or a timeout (mysql_thread___fast_forward_grace_close_ms)
+			 * is reached.
+			 *
+			 * This logic detects backend closure, starts the grace timer, and waits
+			 * for buffers to empty before closing the session.
+			 */
+			// Detect if backend closed during fast forward
+			if (mybe->server_myds->status == MYSQL_SERVER_STATUS_OFFLINE_HARD || mybe->server_myds->fd == -1) {
+				if (!backend_closed_in_fast_forward) {
+					backend_closed_in_fast_forward = true;
+					fast_forward_grace_start_time = thread->curtime;
+				}
+			}
+			if (backend_closed_in_fast_forward) {
+				if (
+					( mybe->server_myds == nullptr || ( mybe->server_myds && mybe->server_myds->PSarrayIN->len == 0 ) )
+					&&
+					(client_myds->PSarrayOUT->len == 0 && (client_myds->queueOUT.head - client_myds->queueOUT.tail) == 0)
+				) {
+					// buffers empty, close
+					handler_ret = -1;
+				} else if (thread->curtime - fast_forward_grace_start_time > (unsigned long long)mysql_thread___fast_forward_grace_close_ms * 1000) {
+					// timeout, close
+					handler_ret = -1;
+				}
+			}
 			break;
 		// This state is required because it covers the following situation:
 		//  1. A new connection is created by a client and the 'FAST_FORWARD' mode is enabled.
diff --git a/lib/MySQL_Thread.cpp b/lib/MySQL_Thread.cpp
@@ -511,6 +511,7 @@ static char * mysql_thread_variables_names[]= {
 	(char *)"proxy_protocol_networks",
 	(char *)"protocol_compression_level",
 	(char *)"ignore_min_gtid_annotations",
+	(char *)"fast_forward_grace_close_ms",
 	NULL
 };
 
@@ -1073,8 +1074,12 @@ MySQL_Threads_Handler::MySQL_Threads_Handler() {
 		variables.default_variables[i]=strdup(mysql_tracked_variables[i].default_value);
 	}
 	variables.default_session_track_gtids=strdup((char *)MYSQL_DEFAULT_SESSION_TRACK_GTIDS);
+    // fast_forward_grace_close_ms: Configurable timeout (in milliseconds) for the "fast forward grace close" feature.
+    // This feature prevents data loss in fast forward mode by deferring session closure when the backend
+    // connection closes unexpectedly, allowing time for pending client output to drain.
 	variables.ping_interval_server_msec=10000;
 	variables.ping_timeout_server=200;
+	variables.fast_forward_grace_close_ms=5000;
 	variables.default_schema=strdup((char *)"information_schema");
 	variables.handle_unknown_charset=1;
 	variables.interfaces=strdup((char *)"");
@@ -2283,6 +2288,7 @@ char ** MySQL_Threads_Handler::get_variables_list() {
 		VariablesPointers_int["handle_unknown_charset"]        = make_tuple(&variables.handle_unknown_charset,        0, HANDLE_UNKNOWN_CHARSET__MAX_HANDLE_VALUE, false);
 		VariablesPointers_int["ping_interval_server_msec"]     = make_tuple(&variables.ping_interval_server_msec,  1000, 7*24*3600*1000, false);
 		VariablesPointers_int["ping_timeout_server"]           = make_tuple(&variables.ping_timeout_server,          10,       600*1000, false);
+		VariablesPointers_int["fast_forward_grace_close_ms"]   = make_tuple(&variables.fast_forward_grace_close_ms,   0,      3600*1000, false);
 		VariablesPointers_int["client_host_cache_size"]        = make_tuple(&variables.client_host_cache_size,        0,      1024*1024, false);
 		VariablesPointers_int["client_host_error_counts"]      = make_tuple(&variables.client_host_error_counts,      0,      1024*1024, false);
 		VariablesPointers_int["handle_warnings"]			   = make_tuple(&variables.handle_warnings,				  0,			  1, false);
@@ -3748,7 +3754,25 @@ bool MySQL_Thread::process_data_on_data_stream(MySQL_Data_Stream *myds, unsigned
 						// if this is a backend without fast_forward, do not set unhealthy: it will be handled by client library
 						if (myds->sess->session_fast_forward) { // if fast forward
 							if (myds->myds_type==MYDS_BACKEND) { // and backend
-								myds->sess->set_unhealthy(); // set unhealthy
+						//		myds->sess->set_unhealthy(); // set unhealthy
+							// Fast Forward Grace Close Logic:
+							// If the backend closed during fast forward mode, we defer session closure to allow
+							// pending client output buffers to drain, preventing data loss.
+							// Detect if backend closed during fast forward
+								if (myds->sess->backend_closed_in_fast_forward == false) {
+									myds->sess->backend_closed_in_fast_forward = true;
+									//cerr << __FILE__ << ":" << __LINE__ << " grace_start_time from " << myds->sess->fast_forward_grace_start_time << " to " << curtime << endl;
+									myds->sess->fast_forward_grace_start_time = curtime;
+								}
+								if (myds->sess->backend_closed_in_fast_forward) {
+									if (myds->PSarrayIN->len == 0 && myds->sess->client_myds->PSarrayOUT->len == 0 && (myds->sess->client_myds->queueOUT.head - myds->sess->client_myds->queueOUT.tail) == 0) {
+										// buffers empty, close
+										myds->sess->set_unhealthy(); // set unhealthy
+									} else if (curtime - myds->sess->fast_forward_grace_start_time > (unsigned long long)mysql_thread___fast_forward_grace_close_ms * 1000) {
+										// timeout, close
+										myds->sess->set_unhealthy(); // set unhealthy
+									}
+								}
 							}
 						}
 					}
@@ -3925,9 +3949,17 @@ void MySQL_Thread::ProcessAllSessions_Healthy0(MySQL_Session *sess, unsigned int
 					sess->client_myds->addr.port
 				);
 			} else {
+				string extra_info = "";
+				if (sess->backend_closed_in_fast_forward == true) {
+					unsigned long long lapse = curtime - sess->fast_forward_grace_start_time;
+					extra_info = "Yes , " + to_string(lapse/1000) + " ms ago"; 
+				} else {
+					extra_info = "No";
+				}
 				proxy_warning(
-					"Closing 'fast_forward' client connection %s:%d\n", sess->client_myds->addr.addr,
-					sess->client_myds->addr.port
+					"Closing 'fast_forward' client connection %s:%d . Backend already close: %s\n",
+					sess->client_myds->addr.addr, sess->client_myds->addr.port,
+					extra_info.c_str()
 				);
 			}
 		}
@@ -4137,6 +4169,7 @@ void MySQL_Thread::refresh_variables() {
 	REFRESH_VARIABLE_INT(connect_timeout_server);
 	REFRESH_VARIABLE_INT(connect_timeout_server_max);
 	REFRESH_VARIABLE_INT(free_connections_pct);
+	REFRESH_VARIABLE_INT(fast_forward_grace_close_ms);
 #ifdef IDLE_THREADS
 	REFRESH_VARIABLE_INT(session_idle_ms);
 #endif // IDLE_THREADS
diff --git a/lib/mysql_data_stream.cpp b/lib/mysql_data_stream.cpp
@@ -576,7 +576,26 @@ int MySQL_Data_Stream::read_from_net() {
 		} else {
 			// Shutdown if we either received the EOF, or operation failed with non-retryable error.
 			if (ssl_recv_bytes==0 || (ssl_recv_bytes==-1 && errno != EINTR && errno != EAGAIN)) {
-				proxy_debug(PROXY_DEBUG_NET, 5, "Received EOF, shutting down soft socket -- Session=%p, Datastream=%p\n", sess, this);
+				/*
+				 * Fast Forward Grace Close Logic:
+				 * When the backend connection closes unexpectedly (EOF) during fast forward mode,
+				 * instead of immediately closing the session, we check if there are pending
+				 * client output buffers. If so, we initiate a grace period to allow the
+				 * buffers to drain before closing the session.
+				 *
+				 * This prevents data loss in fast forward scenarios where ProxySQL forwards
+				 * packets without buffering, and the backend closes before all data is sent.
+				 */
+				if (myds_type == MYDS_BACKEND && sess && sess->session_fast_forward && ssl_recv_bytes==0) {
+					if (PSarrayIN->len > 0 || sess->client_myds->PSarrayOUT->len > 0 || queue_data(sess->client_myds->queueOUT) > 0) {
+						if (sess->backend_closed_in_fast_forward == false) {
+							sess->backend_closed_in_fast_forward = true;
+							sess->fast_forward_grace_start_time = sess->thread->curtime;
+							sess->client_myds->defer_close_due_to_fast_forward = true;
+						}
+					}
+				}
+				proxy_debug(PROXY_DEBUG_NET, 5, "Received EOF, shutting down soft socket -- Session=%p, Datastream=%p", sess, this);
 				shut_soft();
 				return -1;
 			}
@@ -590,6 +609,19 @@ int MySQL_Data_Stream::read_from_net() {
 		if (encrypted==false) {
 			int myds_errno=errno;
 			if (r==0 || (r==-1 && myds_errno != EINTR && myds_errno != EAGAIN)) {
+				/*
+				 * Fast Forward Grace Close Logic:
+				 * Similar check for non-encrypted connections when backend closes with EOF.
+				 */
+				if (myds_type == MYDS_BACKEND && sess && sess->session_fast_forward && r==0) {
+					if (PSarrayIN->len > 0 || sess->client_myds->PSarrayOUT->len > 0 || queue_data(sess->client_myds->queueOUT) > 0) {
+						if (sess->backend_closed_in_fast_forward == false) {
+							sess->backend_closed_in_fast_forward = true;
+							sess->fast_forward_grace_start_time = sess->thread->curtime;
+							sess->client_myds->defer_close_due_to_fast_forward = true;
+						}
+					}
+				}
 				shut_soft();
 			}
 		} else {
@@ -622,7 +654,20 @@ int MySQL_Data_Stream::read_from_net() {
 		if ( (revents & POLLHUP) ) {
 			// this is a final check
 			// Only if the amount of data read is 0 or less, then we check POLLHUP
-			proxy_debug(PROXY_DEBUG_NET, 5, "Session=%p, Datastream=%p -- shutdown soft. revents=%d , bytes read = %d\n", sess, this, revents, r);
+			/*
+			 * Fast Forward Grace Close Logic:
+			 * Handle POLLHUP event similarly, initiating grace close if buffers are pending.
+			 */
+			if (myds_type == MYDS_BACKEND && sess && sess->session_fast_forward) {
+				if (PSarrayIN->len > 0 || sess->client_myds->PSarrayOUT->len > 0 || queue_data(sess->client_myds->queueOUT) > 0) {
+					if (sess->backend_closed_in_fast_forward == false) {
+						sess->backend_closed_in_fast_forward = true;
+						sess->fast_forward_grace_start_time = sess->thread->curtime;
+						sess->client_myds->defer_close_due_to_fast_forward = true;
+					}
+				}
+			}
+			proxy_debug(PROXY_DEBUG_NET, 5, "Session=%p, Datastream=%p -- shutdown soft. revents=%d , bytes read = %d", sess, this, revents, r);
 			shut_soft();
 		}
 	} else {
@@ -765,6 +810,22 @@ void MySQL_Data_Stream::set_pollout() {
 		_pollfd->events = myconn->wait_events;
 	} else {
 		_pollfd->events = POLLIN;
+		if (myds_type == MYDS_BACKEND && sess && sess->session_fast_forward && sess->backend_closed_in_fast_forward == true) {
+			/*
+			 * Fast Forward Grace Close Logic:
+			 * During the grace period after backend closure, we manage polling to avoid busy-waiting.
+			 * If POLLIN is set, poll() will return immediately since the socket is closed,
+			 * causing the thread to spin. To prevent this, we clear POLLIN during the grace period
+			 * and rely on timeouts to eventually close the session.
+			 */
+			// this is a fast forward session where the backend connection was already closed
+			// if we set POLLIN : the thread will spin on poll() until the socket is closed
+			// if we do not set POLLIN : we won't be able to timeout
+			if (sess->thread->curtime - sess->fast_forward_grace_start_time < (unsigned long long)mysql_thread___fast_forward_grace_close_ms * 1000) {
+				// for the reason listed above, we remove POLLIN unless the timeout has reached
+				_pollfd->events = 0;
+			}
+		}
 		//if (PSarrayOUT->len || available_data_out() || queueOUT.partial || (encrypted && !SSL_is_init_finished(ssl))) {
 		if (PSarrayOUT->len || available_data_out() || queueOUT.partial) {
 			_pollfd->events |= POLLOUT;
diff --git a/test/tap/groups/groups.json b/test/tap/groups/groups.json
@@ -137,6 +137,7 @@
   "test_cluster_sync_mysql_servers-t" : [ "default-g3","mysql-auto_increment_delay_multiplex=0-g3","mysql-multiplexing=false-g3","mysql-query_digests=0-g3","mysql-query_digests_keep_comment=1-g3" ],
   "test_cluster_sync-t" : [ "default-g3","mysql-auto_increment_delay_multiplex=0-g3","mysql-multiplexing=false-g3","mysql-query_digests=0-g3","mysql-query_digests_keep_comment=1-g3" ],
   "test_com_binlog_dump_enables_fast_forward-t" : [ "default-g3","mysql-auto_increment_delay_multiplex=0-g3","mysql-multiplexing=false-g3","mysql-query_digests=0-g3","mysql-query_digests_keep_comment=1-g3" ],
+  "fast_forward_grace_close_libmysql-t" : [ "default-g3","mysql-auto_increment_delay_multiplex=0-g3","mysql-multiplexing=false-g3","mysql-query_digests=0-g3","mysql-query_digests_keep_comment=1-g3" ],
   "test_com_register_slave_enables_fast_forward-t" : [ "default-g3","mysql-auto_increment_delay_multiplex=0-g3","mysql-multiplexing=false-g3","mysql-query_digests=0-g3","mysql-query_digests_keep_comment=1-g3" ],
   "test_com_reset_connection_com_change_user-t" : [ "default-g3","mysql-auto_increment_delay_multiplex=0-g3","mysql-multiplexing=false-g3","mysql-query_digests=0-g3","mysql-query_digests_keep_comment=1-g3" ],
   "test_connection_annotation-t" : [ "default-g3","mysql-auto_increment_delay_multiplex=0-g3","mysql-multiplexing=false-g3","mysql-query_digests=0-g3","mysql-query_digests_keep_comment=1-g3" ],
diff --git a/test/tap/tests/Makefile b/test/tap/tests/Makefile
@@ -121,6 +121,7 @@ tests: tests-cpp \
 	reg_test_stmt_resultset_err_no_rows_libmysql-t \
 	prepare_statement_err3024_libmysql-t \
 	prepare_statement_err3024_async-t \
+	fast_forward_grace_close_libmysql-t \
 	reg_test_mariadb_stmt_store_result_libmysql-t \
 	reg_test_mariadb_stmt_store_result_async-t
 tests:
@@ -229,6 +230,9 @@ mysql_reconnect_libmariadb-t: mysql_reconnect.cpp $(TAP_LDIR)/libtap.so
 mysql_reconnect_libmysql-t: mysql_reconnect.cpp $(TAP_LDIR)/libtap_mysql8.a
 	$(CXX) -DLIBMYSQL_HELPER8 -DDISABLE_WARNING_COUNT_LOGGING $< -I$(TEST_MYSQL8_IDIR) -I$(TEST_MYSQL8_EDIR) -L$(TEST_MYSQL8_LDIR) -lmysqlclient -ltap_mysql8 -lresolv $(CUSTOMARGS) -o $@
 
+fast_forward_grace_close_libmysql-t: fast_forward_grace_close.cpp $(TAP_LDIR)/libtap_mysql8.a
+	$(CXX) -DLIBMYSQL_HELPER8 -DDISABLE_WARNING_COUNT_LOGGING $< -I$(TEST_MYSQL8_IDIR) -I$(TEST_MYSQL8_EDIR) -L$(TEST_MYSQL8_LDIR) -lmysqlclient -ltap_mysql8 -lresolv $(CUSTOMARGS) -o $@
+
 test_match_eof_conn_cap_libmysql-t: test_match_eof_conn_cap.cpp $(TAP_LDIR)/libtap_mysql8.a
 	$(CXX) -DLIBMYSQL_HELPER8 -DDISABLE_WARNING_COUNT_LOGGING $< -I$(TEST_MYSQL8_IDIR) -I$(TEST_MYSQL8_EDIR) -L$(TEST_MYSQL8_LDIR) -lmysqlclient -ltap_mysql8 -lresolv $(CUSTOMARGS) -o $@
 
diff --git a/test/tap/tests/fast_forward_grace_close.cpp b/test/tap/tests/fast_forward_grace_close.cpp
diff --git a/test/tap/tests/test_com_binlog_dump_enables_fast_forward-t.cpp b/test/tap/tests/test_com_binlog_dump_enables_fast_forward-t.cpp
