Merge branch 'v3.0'

Makefile

@@ -314,7 +314,7 @@ amd64-fedora: fedora38 fedora38-clang fedora38-dbg fedora39 fedora39-clang fedor
 amd64-opensuse: opensuse15 opensuse15-clang opensuse15-dbg
 amd64-ubuntu: ubuntu18 ubuntu18-dbg ubuntu20 ubuntu20-clang ubuntu20-dbg ubuntu22 ubuntu22-clang ubuntu22-dbg ubuntu24 ubuntu24-clang ubuntu24-dbg
 amd64-pkglist:
-	@make -nk amd64-packages 2>/dev/null | grep -Po '(?<=binaries/)proxysql\S+'
+	@make -nk amd64-packages 2>/dev/null | grep -Po '(?<=binaries/)proxysql\S+$$'
 
 arm64-%: SYS_ARCH := aarch64
 arm64-packages: arm64-centos arm64-debian arm64-ubuntu arm64-fedora arm64-opensuse arm64-almalinux
@@ -325,7 +325,7 @@ arm64-fedora: fedora38 fedora39 fedora40 fedora41
 arm64-opensuse: opensuse15
 arm64-ubuntu: ubuntu18 ubuntu20 ubuntu22 ubuntu24
 arm64-pkglist:
-	@make -nk arm64-packages 2>/dev/null | grep -Po '(?<=binaries/)proxysql\S+'
+	@make -nk arm64-packages 2>/dev/null | grep -Po '(?<=binaries/)proxysql\S+$$'
 
 almalinux%: build-almalinux% ;
 centos%: build-centos% ;

deps/Makefile

@@ -237,6 +237,7 @@ endif
 	# patches for replication testing
 	cd mariadb-client-library/mariadb_client && patch -p0 < ../mariadb_rpl.patch
 	cd mariadb-client-library/mariadb_client && patch -p0 < ../cmakelists.txt.patch
+	cd mariadb-client-library/mariadb_client && patch -p0 < ../mariadb_lib.c.metadata_column_check.patch
 	cd mariadb-client-library/mariadb_client && CC=${CC} CXX=${CXX} ${MAKE} mariadbclient
 #	cd mariadb-client-library/mariadb_client/include && make my_config.h
 

deps/libscram/src/scram.c

@@ -43,7 +43,7 @@ void scram_reset_error() {
 	errorBuffer[0] = '\0';
 }
 
-static size_t strlcat(char* dst, const char* src, size_t siz)
+static size_t my_strlcat(char* dst, const char* src, size_t siz)
 {
 	char* d = dst;
 	const char* s = src;
@@ -434,7 +434,7 @@ char *build_client_final_message(ScramState *scram_state,
 				    client_proof))
 		goto failed;
 
-	len = strlcat(buf, ",p=", sizeof(buf));
+	len = my_strlcat(buf, ",p=", sizeof(buf));
 	enclen = pg_b64_enc_len(sizeof(client_proof));
 	enclen = pg_b64_encode((char *) client_proof,
 			       SCRAM_KEY_LEN,
@@ -597,15 +597,15 @@ static bool calculate_client_proof(ScramState *scram_state,
 								 &errstr) < 0 ||
 			scram_ClientKey(scram_state->SaltedPassword, PG_SHA256, SCRAM_KEY_LEN, ClientKey, 
 						    &errstr) < 0) {
-			snprintf(errorBuffer, MAX_ERROR_LENGTH, "could not compute client key: %s", *errstr);
+			snprintf(errorBuffer, MAX_ERROR_LENGTH, "could not compute client key: %s", (const char*) errstr);
 			goto failed;
 		}
 	}
 
 	/* Hash it one more time, and compare with StoredKey */
 	if (scram_H(ClientKey, PG_SHA256, SCRAM_KEY_LEN,
 		StoredKey, &errstr) < 0) {
-		snprintf(errorBuffer, MAX_ERROR_LENGTH, "could not hash stored key: %s", *errstr);
+		snprintf(errorBuffer, MAX_ERROR_LENGTH, "could not hash stored key: %s", (const char*) errstr);
 		goto failed;
 	}
 	/*
@@ -656,7 +656,7 @@ bool verify_server_signature(ScramState *scram_state, const PgCredentials *crede
 		memcpy(ServerKey, credentials->scram_ServerKey, SCRAM_KEY_LEN);
 	else {
 		if (scram_ServerKey(scram_state->SaltedPassword, PG_SHA256, SCRAM_KEY_LEN, ServerKey, &errstr) < 0) {
-			snprintf(errorBuffer, MAX_ERROR_LENGTH, "could not compute server key: %s", *errstr);
+			snprintf(errorBuffer, MAX_ERROR_LENGTH, "could not compute server key: %s", (const char*) errstr);
 			goto failed;
 		}
 	}
@@ -907,17 +907,17 @@ static bool build_adhoc_scram_secret(const char *plain_password, ScramState *scr
 			     scram_state->iterations,
 			     salted_password, &errstr) < 0 ||
 		scram_ClientKey(salted_password, PG_SHA256, SCRAM_KEY_LEN, scram_state->StoredKey, &errstr) < 0) {
-		snprintf(errorBuffer, MAX_ERROR_LENGTH, "could not compute server key: %s", *errstr);
+		snprintf(errorBuffer, MAX_ERROR_LENGTH, "could not compute server key: %s", (const char*) errstr);
 		goto failed;
 	}
 
 	if (scram_H(scram_state->StoredKey, PG_SHA256, SCRAM_KEY_LEN, scram_state->StoredKey, &errstr) < 0) {
-		snprintf(errorBuffer, MAX_ERROR_LENGTH, "could not hash stored key: %s", *errstr);
+		snprintf(errorBuffer, MAX_ERROR_LENGTH, "could not hash stored key: %s", (const char*) errstr);
 		goto failed;
 	}
 
 	if (scram_ServerKey(salted_password, PG_SHA256, SCRAM_KEY_LEN, scram_state->ServerKey, &errstr) < 0) {
-		snprintf(errorBuffer, MAX_ERROR_LENGTH, "could not calculate server key: %s", *errstr);
+		snprintf(errorBuffer, MAX_ERROR_LENGTH, "could not calculate server key: %s", (const char*) errstr);
 		goto failed;
 	}
 
@@ -1209,7 +1209,7 @@ bool verify_client_proof(ScramState *state, const char *ClientProof)
 
 	/* Hash it one more time, and compare with StoredKey */
 	if (scram_H(state->ClientKey, PG_SHA256, SCRAM_KEY_LEN, client_StoredKey, &errstr) < 0) {
-		snprintf(errorBuffer, MAX_ERROR_LENGTH, "could not hash stored key: %s", *errstr);
+		snprintf(errorBuffer, MAX_ERROR_LENGTH, "could not hash stored key: %s", (const char*) errstr);
 		goto failed;
 	}
 
@@ -1271,7 +1271,7 @@ bool scram_verify_plain_password(const char *username, const char *password,
 	/* Compute Server Key based on the user-supplied plaintext password */
 	if (scram_SaltedPassword(password, PG_SHA256, SCRAM_KEY_LEN, salt, saltlen, iterations, salted_password, &errstr) < 0 ||
 		scram_ServerKey(salted_password, PG_SHA256, SCRAM_KEY_LEN, computed_key, &errstr) < 0) {
-		snprintf(errorBuffer, MAX_ERROR_LENGTH, "could not compute server key: %s", *errstr);
+		snprintf(errorBuffer, MAX_ERROR_LENGTH, "could not compute server key: %s", (const char*) errstr);
 		goto failed;
 	}
 

deps/mariadb-client-library/mariadb_lib.c.metadata_column_check.patch

@@ -0,0 +1,31 @@
+diff --git libmariadb/mariadb_lib.c libmariadb/mariadb_lib.c
+index 027167f1..58b8283a 100644
+--- libmariadb/mariadb_lib.c
++++ libmariadb/mariadb_lib.c
+@@ -3021,6 +3021,12 @@ MYSQL_FIELD *ma_duplicate_resultset_metadata(MYSQL_FIELD *fields, size_t count,
+   return result;
+ }
+
++static uint8_t mysql_encode_length(uint64_t len) {
++    if (len < 251) { return 1; }
++    if (len < 65536) { return 3; }
++    if (len < 16777216) { return 4; }
++    return 9;
++}
+
+ int mthd_my_read_query_result(MYSQL *mysql)
+ {
+@@ -3070,6 +3076,13 @@ get_info:
+
+   if (has_metadata)
+   {
++    // integrity-check: the length encoding of the field count from 'column-count' packet
++    // must match the packet length from header, otherwise packet is malformed.
++    ulong enc_len = mysql_encode_length(field_count);
++    if (enc_len != length) {
++      my_set_error(mysql, CR_MALFORMED_PACKET, SQLSTATE_UNKNOWN, 0);
++      return -1;
++    }
+     // read packet metadata
+     mysql->fields =
+         mthd_my_read_metadata(mysql, field_count, 7 + ma_extended_type_info_rows(mysql));

include/MySQL_Monitor.hpp

@@ -436,7 +436,8 @@ struct DNS_Resolve_Data {
 	std::shared_ptr<DNS_Cache> dns_cache;
 	std::string hostname;
 	std::set<std::string> cached_ips;
-	unsigned int ttl;
+	unsigned int ttl = 0;
+	unsigned int refresh_intv = 0;
 };
 
 

include/MySQL_Query_Cache.h

@@ -0,0 +1,26 @@
+#ifndef __CLASS_MYSQL_QUERY_CACHE_H
+#define __CLASS_MYSQL_QUERY_CACHE_H
+
+#include "proxysql.h"
+#include "cpp.h"
+#include "query_cache.hpp"
+
+typedef struct _MySQL_QC_entry : public QC_entry_t {
+	uint32_t column_eof_pkt_offset;
+	uint32_t row_eof_pkt_offset;
+	uint32_t ok_pkt_offset;
+} MySQL_QC_entry_t;
+
+class MySQL_Query_Cache : public Query_Cache<MySQL_Query_Cache> {
+public:
+	MySQL_Query_Cache() = default;
+	~MySQL_Query_Cache() = default;
+
+	bool set(uint64_t user_hash, const unsigned char* kp, uint32_t kl, unsigned char* vp, uint32_t vl, 
+		uint64_t create_ms, uint64_t curtime_ms, uint64_t expire_ms, bool deprecate_eof_active);
+	unsigned char* get(uint64_t user_hash, const unsigned char* kp, const uint32_t kl, uint32_t* lv, 
+		uint64_t curtime_ms, uint64_t cache_ttl, bool deprecate_eof_active);
+	//void* purgeHash_thread(void*);
+};
+
+#endif /* __CLASS_MYSQL_QUERY_CACHE_H */

include/PgSQL_Data_Stream.h

@@ -107,8 +107,8 @@ class PgSQL_Data_Stream
 	FixedSizeQueue data_packets_history_IN;
 	FixedSizeQueue data_packets_history_OUT;
 	//PtrSizeArray *PSarrayOUTpending;
-	PtrSizeArray* resultset;
-	unsigned int resultset_length;
+	//PtrSizeArray* resultset;
+	//unsigned int resultset_length;
 
 	ProxySQL_Poll<PgSQL_Data_Stream>* mypolls;
 	//int listener;
@@ -200,8 +200,9 @@ class PgSQL_Data_Stream
 	void check_data_flow();
 	int assign_fd_from_mysql_conn();
 
-	unsigned char* resultset2buffer(bool);
-	void buffer2resultset(unsigned char*, unsigned int);
+	static unsigned char* copy_array_to_buffer(PtrSizeArray* resultset, size_t resultset_length, bool del);
+	static void copy_buffer_to_resultset(PtrSizeArray* resultset, unsigned char* ptr, uint64_t size, 
+		char current_transaction_state);
 
 	// safe way to attach a PgSQL Connection
 	void attach_connection(PgSQL_Connection* mc) {

include/PgSQL_HostGroups_Manager.h

@@ -838,7 +838,7 @@ class PgSQL_HostGroups_Manager : public Base_HostGroups_Manager<PgSQL_HGC> {
 	void replication_lag_action_inner(PgSQL_HGC *, const char*, unsigned int, int);
 	void replication_lag_action(const std::list<replication_lag_server_t>& pgsql_servers);
 	void read_only_action(char *hostname, int port, int read_only);
-	void read_only_action_v2(const std::list<read_only_server_t>& pgsql_servers);
+	void read_only_action_v2(const std::list<read_only_server_t>& pgsql_servers, bool writer_is_also_reader);
 	unsigned int get_servers_table_version();
 	void wait_servers_table_version(unsigned, unsigned);
 	bool shun_and_killall(char *hostname, int port);

include/PgSQL_Monitor.hpp

@@ -14,6 +14,8 @@
 
 #define MONITOR_SQLITE_TABLE_PGSQL_SERVER_PING_LOG "CREATE TABLE pgsql_server_ping_log ( hostname VARCHAR NOT NULL , port INT NOT NULL DEFAULT 3306 , time_start_us INT NOT NULL DEFAULT 0 , ping_success_time_us INT DEFAULT 0 , ping_error VARCHAR , PRIMARY KEY (hostname, port, time_start_us))"
 
+#define MONITOR_SQLITE_TABLE_PGSQL_SERVER_READ_ONLY_LOG "CREATE TABLE pgsql_server_read_only_log ( hostname VARCHAR NOT NULL , port INT NOT NULL DEFAULT 3306 , time_start_us INT NOT NULL DEFAULT 0 , success_time_us INT DEFAULT 0 , read_only INT DEFAULT 1 , error VARCHAR , PRIMARY KEY (hostname, port, time_start_us))"
+
 #define MONITOR_SQLITE_TABLE_PGSQL_SERVERS "CREATE TABLE pgsql_servers (hostname VARCHAR NOT NULL , port INT NOT NULL DEFAULT 3306 , status INT CHECK (status IN (0, 1, 2, 3, 4)) NOT NULL DEFAULT 0 , use_ssl INT CHECK (use_ssl IN(0,1)) NOT NULL DEFAULT 0 , PRIMARY KEY (hostname, port) )"
 
 #define MONITOR_SQLITE_TABLE_PROXYSQL_SERVERS "CREATE TABLE proxysql_servers (hostname VARCHAR NOT NULL , port INT NOT NULL DEFAULT 6032 , weight INT CHECK (weight >= 0) NOT NULL DEFAULT 0 , comment VARCHAR NOT NULL DEFAULT '' , PRIMARY KEY (hostname, port) )"
@@ -35,6 +37,8 @@ struct PgSQL_Monitor {
 	uint64_t connect_check_OK { 0 };
 	uint64_t ping_check_ERR { 0 };
 	uint64_t ping_check_OK { 0 };
+	uint64_t readonly_check_ERR { 0 };
+	uint64_t readonly_check_OK { 0 };
 	///////////////////////////////////////////////////////////////////////////
 
 	std::vector<table_def_t> tables_defs_monitor {
@@ -45,7 +49,11 @@ struct PgSQL_Monitor {
 		{
 			const_cast<char*>("pgsql_server_ping_log"),
 			const_cast<char*>(MONITOR_SQLITE_TABLE_PGSQL_SERVER_PING_LOG)
-		}
+		},
+		{
+			const_cast<char*>("pgsql_server_read_only_log"),
+			const_cast<char*>(MONITOR_SQLITE_TABLE_PGSQL_SERVER_READ_ONLY_LOG)
+		},
 	};
 
 	std::vector<table_def_t> tables_defs_monitor_internal {

include/PgSQL_Query_Cache.h

@@ -0,0 +1,22 @@
+#ifndef __CLASS_PGSQL_QUERY_CACHE_H
+#define __CLASS_PGSQL_QUERY_CACHE_H
+
+#include "proxysql.h"
+#include "cpp.h"
+#include "query_cache.hpp"
+
+typedef struct _PgSQL_QC_entry : public QC_entry_t {} PgSQL_QC_entry_t;
+
+class PgSQL_Query_Cache : public Query_Cache<PgSQL_Query_Cache> {
+public:
+	PgSQL_Query_Cache() = default;
+	~PgSQL_Query_Cache() = default;
+
+	bool set(uint64_t user_hash, const unsigned char* kp, uint32_t kl, unsigned char* vp, uint32_t vl, 
+		uint64_t create_ms, uint64_t curtime_ms, uint64_t expire_ms);
+	const std::shared_ptr<PgSQL_QC_entry_t> get(uint64_t user_hash, const unsigned char* kp, const uint32_t kl, 
+		uint64_t curtime_ms, uint64_t cache_ttl);
+	//void* purgeHash_thread(void*);
+};
+
+#endif /* __CLASS_PGSQL_QUERY_CACHE_H */

include/PgSQL_Thread.h

@@ -820,16 +820,19 @@ class PgSQL_Threads_Handler
 
 		int monitor_history;
 		int monitor_connect_interval;
+		int monitor_connect_interval_window;
 		int monitor_connect_timeout;
 		//! Monitor ping interval. Unit: 'ms'.
 		int monitor_ping_interval;
+		int monitor_ping_interval_window;
 		int monitor_ping_max_failures;
 		//! Monitor ping timeout. Unit: 'ms'.
 		int monitor_ping_timeout;
 		//! Monitor aws rds topology discovery interval. Unit: 'one discovery check per X monitor_read_only checks'.
 		int monitor_aws_rds_topology_discovery_interval;
 		//! Monitor read only timeout. Unit: 'ms'.
 		int monitor_read_only_interval;
+		int monitor_read_only_interval_window;
 		//! Monitor read only timeout. Unit: 'ms'.
 		int monitor_read_only_timeout;
 		int monitor_read_only_max_timeout_count;
@@ -867,6 +870,7 @@ class PgSQL_Threads_Handler
 		int monitor_local_dns_resolver_queue_maxsize;
 		char* monitor_username;
 		char* monitor_password;
+		char* monitor_dbname;
 		char* monitor_replication_lag_use_percona_heartbeat;
 		int ping_interval_server_msec;
 		int ping_timeout_server;

include/cpp.h

@@ -10,7 +10,7 @@
 #include "PgSQL_Backend.h"
 #include "ProxySQL_Poll.h"
 //#include "MySQL_Data_Stream.h"
-#include "query_cache.hpp"
+//#include "MySQL_Query_Cache.h"
 #include "mysql_connection.h"
 #include "sqlite3db.h"
 //#include "StatCounters.h"

include/gen_utils.h

@@ -314,6 +314,18 @@ inline unsigned long long realtime_time() {
   return (((unsigned long long) ts.tv_sec) * 1000000) + (ts.tv_nsec / 1000);
 }
 
+template<int FACTOR, typename T>
+inline T overflow_safe_multiply(T val) {
+	static_assert(std::is_integral<T>::value, "T must be an integer type.");
+	static_assert(std::is_unsigned_v<T>, "T must be an unsigned integer type.");
+	static_assert(FACTOR > 0, "Negative factors are not supported.");
+
+	if constexpr (FACTOR == 0) return 0;
+	if (val == 0)  return 0;
+	if (val > std::numeric_limits<T>::max() / FACTOR) return std::numeric_limits<T>::max();
+	return (val * FACTOR);
+}
+
 #endif /* __GEN_FUNCTIONS */
 
 bool Proxy_file_exists(const char *);

include/proxysql_glovars.hpp

@@ -86,7 +86,8 @@ class ProxySQL_GlobalVariables {
 		unsigned long long start_time;
 		bool gdbg;
 		bool nostart;
-		bool monitor;
+		bool my_monitor;
+		bool pg_monitor;
 		bool version_check;
 #ifdef SO_REUSEPORT
 		bool reuseport;