fix(MongoDB): Add timeout support and harden replica set connections (#5204)

* fix(MongoDB): Add timeouts to network connections to database to prevent hangs when servers are not accessible.

* enh(MongoDB): Many document consistency checks and improvements.

* enh(MongoDB): Improved handling of monodb replica sets by using last write reported from hello.

Author: matejk

MongoDB/include/Poco/MongoDB/BSONReader.h

@@ -20,6 +20,7 @@
 
 #include "Poco/MongoDB/MongoDB.h"
 #include "Poco/BinaryReader.h"
+#include "Poco/Exception.h"
 
 
 namespace Poco {
@@ -72,6 +73,8 @@ inline std::string BSONReader::readCString()
 		{
 			if (c == 0x00) return val;
 			else val += c;
+			if (val.size() > static_cast<std::size_t>(BSON_MAX_DOCUMENT_SIZE))
+				throw Poco::DataFormatException("BSON cstring exceeds maximum size");
 		}
 	}
 	return val;

MongoDB/include/Poco/MongoDB/Binary.h

@@ -154,13 +154,19 @@ inline void BSONReader::read<Binary::Ptr>(Binary::Ptr& to)
 	Poco::Int32 size;
 	_reader >> size;
 
+	if (size < 0)
+		throw Poco::DataFormatException("Invalid BSON binary size: " + std::to_string(size));
+	if (size > BSON_MAX_DOCUMENT_SIZE)
+		throw Poco::DataFormatException("BSON binary size exceeds maximum: " + std::to_string(size));
+
 	to->buffer().resize(size);
 
 	unsigned char subtype;
 	_reader >> subtype;
 	to->subtype(subtype);
 
-	_reader.readRaw(reinterpret_cast<char*>(to->buffer().begin()), size);
+	if (size > 0)
+		_reader.readRaw(reinterpret_cast<char*>(to->buffer().begin()), size);
 }
 
 

MongoDB/include/Poco/MongoDB/Connection.h

@@ -136,6 +136,13 @@ class MongoDB_API Connection
 	void connect(const Poco::Net::SocketAddress& addrs);
 		/// Connects to the given MongoDB server.
 
+	void connect(const Poco::Net::SocketAddress& addrs, const Poco::Timespan& connectTimeout, const Poco::Timespan& socketTimeout = 0);
+		/// Connects to the given MongoDB server with the specified connect timeout.
+		/// If connectTimeout is non-zero, the connection attempt will be aborted
+		/// after the specified time.
+		/// If socketTimeout is non-zero, the send and receive timeouts on the socket
+		/// will be set after a successful connection.
+
 	void connect(const Poco::Net::StreamSocket& socket);
 		/// Connects using an already connected socket.
 

MongoDB/include/Poco/MongoDB/Element.h

@@ -21,6 +21,7 @@
 #include "Poco/BinaryReader.h"
 #include "Poco/BinaryWriter.h"
 #include "Poco/DateTimeFormatter.h"
+#include "Poco/Exception.h"
 #include "Poco/MongoDB/BSONReader.h"
 #include "Poco/MongoDB/BSONWriter.h"
 #include "Poco/MongoDB/MongoDB.h"
@@ -188,6 +189,10 @@ inline void BSONReader::read<std::string>(std::string& to)
 {
 	Poco::Int32 size;
 	_reader >> size;
+	if (size < BSON_MIN_STRING_SIZE)
+		throw Poco::DataFormatException("Invalid BSON string size: " + std::to_string(size));
+	if (size > BSON_MAX_DOCUMENT_SIZE)
+		throw Poco::DataFormatException("BSON string size exceeds maximum: " + std::to_string(size));
 	_reader.readRaw(size, to);
 	to.erase(to.end() - 1); // remove terminating 0
 }

MongoDB/include/Poco/MongoDB/MessageHeader.h

@@ -110,7 +110,7 @@ inline Int32 MessageHeader::getMessageLength() const
 
 inline void MessageHeader::setMessageLength(Int32 length)
 {
-	poco_assert (_messageLength >= 0);
+	poco_assert (length >= 0);
 	_messageLength = MSG_HEADER_SIZE + length;
 }
 

MongoDB/include/Poco/MongoDB/MongoDB.h

@@ -51,6 +51,33 @@
 #endif
 
 
+// MongoDB wire protocol and BSON specification limits.
+// These constants are defined centrally so that all validation
+// code in the library uses the same values.
+
+namespace Poco {
+namespace MongoDB {
+
+/// Maximum BSON document size (16 MB) per MongoDB specification.
+/// Applies to documents, strings, binary data, and cstrings.
+static constexpr Poco::Int32 BSON_MAX_DOCUMENT_SIZE = 16 * 1024 * 1024;
+
+/// Minimum BSON document size (5 bytes): 4-byte size field + 1-byte null terminator.
+static constexpr Poco::Int32 BSON_MIN_DOCUMENT_SIZE = 5;
+
+/// Minimum BSON string size (1 byte for the null terminator).
+static constexpr Poco::Int32 BSON_MIN_STRING_SIZE = 1;
+
+/// Maximum OP_MSG message size (48 MB) per MongoDB specification.
+static constexpr Poco::Int32 OP_MSG_MAX_SIZE = 48 * 1024 * 1024;
+
+/// Default local threshold for "nearest" read preference (15 ms = 15000 µs).
+/// Servers within this threshold of the minimum RTT are eligible for selection.
+static constexpr Poco::Int64 DEFAULT_LOCAL_THRESHOLD_US = 15000;
+
+} } // namespace Poco::MongoDB
+
+
 //
 // Automatically link MongoDB library.
 //

MongoDB/include/Poco/MongoDB/ObjectId.h

@@ -92,12 +92,11 @@ class MongoDB_API ObjectId
 //
 inline Timestamp ObjectId::timestamp() const noexcept
 {
-	int time;
-	char* T = reinterpret_cast<char*>(&time);
-	T[0] = _id[3];
-	T[1] = _id[2];
-	T[2] = _id[1];
-	T[3] = _id[0];
+	const Poco::Int32 time =
+		(static_cast<Poco::Int32>(_id[0]) << 24) |
+		(static_cast<Poco::Int32>(_id[1]) << 16) |
+		(static_cast<Poco::Int32>(_id[2]) <<  8) |
+		 static_cast<Poco::Int32>(_id[3]);
 	return Timestamp::fromEpochTime(static_cast<time_t>(time));
 }
 

MongoDB/include/Poco/MongoDB/PoolableConnectionFactory.h

@@ -20,6 +20,7 @@
 
 #include "Poco/MongoDB/Connection.h"
 #include "Poco/ObjectPool.h"
+#include "Poco/Timespan.h"
 
 
 namespace Poco {
@@ -32,6 +33,9 @@ class PoolableObjectFactory<MongoDB::Connection, MongoDB::Connection::Ptr>
 	///
 	/// If a Connection::SocketFactory is given, it must live for the entire
 	/// lifetime of the PoolableObjectFactory.
+	///
+	/// It is strongly recommended to use one of the timeout-aware constructors
+	/// to avoid indefinite hangs when the server is unreachable.
 {
 public:
 	PoolableObjectFactory(Net::SocketAddress& address):
@@ -46,6 +50,22 @@ class PoolableObjectFactory<MongoDB::Connection, MongoDB::Connection::Ptr>
 	{
 	}
 
+	PoolableObjectFactory(Net::SocketAddress& address, Poco::Timespan connectTimeout, Poco::Timespan socketTimeout = 0):
+		_address(address),
+		_pSocketFactory(nullptr),
+		_connectTimeout(connectTimeout),
+		_socketTimeout(socketTimeout)
+	{
+	}
+
+	PoolableObjectFactory(const std::string& address, Poco::Timespan connectTimeout, Poco::Timespan socketTimeout = 0):
+		_address(address),
+		_pSocketFactory(nullptr),
+		_connectTimeout(connectTimeout),
+		_socketTimeout(socketTimeout)
+	{
+	}
+
 	PoolableObjectFactory(const std::string& uri, MongoDB::Connection::SocketFactory& socketFactory):
 		_uri(uri),
 		_pSocketFactory(&socketFactory)
@@ -55,9 +75,15 @@ class PoolableObjectFactory<MongoDB::Connection, MongoDB::Connection::Ptr>
 	MongoDB::Connection::Ptr createObject()
 	{
 		if (_pSocketFactory)
+		{
 			return new MongoDB::Connection(_uri, *_pSocketFactory);
+		}
 		else
-			return new MongoDB::Connection(_address);
+		{
+			MongoDB::Connection::Ptr conn = new MongoDB::Connection();
+			conn->connect(_address, _connectTimeout, _socketTimeout);
+			return conn;
+		}
 	}
 
 	bool validateObject(MongoDB::Connection::Ptr pObject)
@@ -80,7 +106,9 @@ class PoolableObjectFactory<MongoDB::Connection, MongoDB::Connection::Ptr>
 private:
 	Net::SocketAddress _address;
 	std::string _uri;
-	MongoDB::Connection::SocketFactory* _pSocketFactory;
+	MongoDB::Connection::SocketFactory* _pSocketFactory = nullptr;
+	Poco::Timespan _connectTimeout;
+	Poco::Timespan _socketTimeout;
 };
 
 

MongoDB/include/Poco/MongoDB/ReadPreference.h

@@ -111,8 +111,14 @@ class MongoDB_API ReadPreference
 	[[nodiscard]] Poco::Int64 maxStalenessSeconds() const;
 		/// Returns the max staleness in seconds, or NO_MAX_STALENESS if not set.
 
-	[[nodiscard]] std::vector<ServerDescription> selectServers(const TopologyDescription& topology) const;
+	[[nodiscard]] std::vector<ServerDescription> selectServers(
+		const TopologyDescription& topology,
+		Poco::Int64 heartbeatFrequencyUs = 10000000) const;
 		/// Selects eligible servers from the topology based on this read preference.
+		/// heartbeatFrequencyUs is the topology monitoring interval in microseconds,
+		/// used in the maxStalenessSeconds calculation per the MongoDB Server Selection spec.
+		/// See: https://github.com/mongodb/specifications/blob/master/source/server-selection/server-selection.md
+		/// Defaults to 10 seconds (10000000 us) per the MongoDB specification.
 		/// Returns a vector of eligible servers.
 		/// If no servers match, returns an empty vector.
 
@@ -137,7 +143,8 @@ class MongoDB_API ReadPreference
 private:
 	bool matchesTags(const ServerDescription& server) const;
 	std::vector<ServerDescription> filterByTags(const std::vector<ServerDescription>& servers) const;
-	std::vector<ServerDescription> filterByMaxStaleness(const std::vector<ServerDescription>& servers, const ServerDescription& primary) const;
+	std::vector<ServerDescription> filterByMaxStaleness(const std::vector<ServerDescription>& servers,
+		const ServerDescription& primary, Poco::Int64 heartbeatFrequencyUs) const;
 	std::vector<ServerDescription> selectByNearest(const std::vector<ServerDescription>& servers) const;
 
 	Mode _mode{Primary};

MongoDB/include/Poco/MongoDB/ReplicaSet.h

@@ -23,8 +23,10 @@
 #include "Poco/MongoDB/ReadPreference.h"
 #include "Poco/MongoDB/TopologyDescription.h"
 #include "Poco/Net/SocketAddress.h"
+#include "Poco/Random.h"
 #include <atomic>
 #include <chrono>
+#include <condition_variable>
 #include <cstddef>
 #include <mutex>
 #include <string>
@@ -93,22 +95,20 @@ class MongoDB_API ReplicaSet
 			/// Default read preference for this replica set.
 
 		unsigned int connectTimeoutSeconds{10};
-			/// Connection timeout in seconds (default: 10)
+			/// Connection timeout in seconds (default: 10).
 			///
-			/// NOTE: This value is currently unused by ReplicaSet itself. It is intended
-			/// for use by custom SocketFactory implementations. Custom factories can
-			/// access this value via ReplicaSet::configuration() and use it when creating
-			/// sockets. Use ReplicaSet::setSocketFactory() to set a custom factory that
-			/// utilizes this timeout value.
+			/// Applied when connecting to MongoDB servers during topology monitoring
+			/// and when creating connections via getConnection()/getPrimaryConnection()/
+			/// getSecondaryConnection(). When using a custom SocketFactory, the factory
+			/// is responsible for applying its own connect timeout.
 
 		unsigned int socketTimeoutSeconds{30};
-			/// Socket send/receive timeout in seconds (default: 30)
+			/// Socket send/receive timeout in seconds (default: 30).
 			///
-			/// NOTE: This value is currently unused by ReplicaSet itself. It is intended
-			/// for use by custom SocketFactory implementations. Custom factories can
-			/// access this value via ReplicaSet::configuration() and use it when creating
-			/// sockets. Use ReplicaSet::setSocketFactory() to set a custom factory that
-			/// utilizes this timeout value.
+			/// Applied as the send and receive timeout on sockets after successful
+			/// connection. This affects how long sendRequest()/readResponse() will
+			/// wait before timing out. When using a custom SocketFactory, the factory
+			/// is responsible for applying its own socket timeout.
 
 		unsigned int heartbeatFrequencySeconds{10};
 			/// Topology monitoring interval in seconds (default: 10)
@@ -183,16 +183,33 @@ class MongoDB_API ReplicaSet
 
 	Connection::Ptr getConnection(const ReadPreference& readPref);
 		/// Returns a connection to a server matching the read preference.
+		/// Uses timeouts from Config.
+		/// Returns null if no suitable server is available.
+
+	Connection::Ptr getConnection(const ReadPreference& readPref,
+		const Poco::Timespan& connectTimeout, const Poco::Timespan& socketTimeout);
+		/// Returns a connection to a server matching the read preference,
+		/// using the specified connect and socket timeouts instead of
+		/// the values from Config.
 		/// Returns null if no suitable server is available.
 
 	Connection::Ptr waitForServerAvailability(const ReadPreference& readPref);
 		/// Waits for a server to become available for the given read preference.
+		/// Uses timeouts from Config.
 		/// This method coordinates waiting between multiple threads - only one thread
 		/// performs the actual sleep and topology refresh, while others benefit from
 		/// the refresh done by the first thread.
 		/// Returns a connection if a server becomes available, or null if still unavailable.
 		/// Thread-safe: uses internal synchronization to prevent redundant refresh attempts.
 
+	Connection::Ptr waitForServerAvailability(const ReadPreference& readPref,
+		const Poco::Timespan& connectTimeout, const Poco::Timespan& socketTimeout);
+		/// Waits for a server to become available for the given read preference,
+		/// using the specified connect and socket timeouts instead of
+		/// the values from Config.
+		/// Returns a connection if a server becomes available, or null if still unavailable.
+		/// Thread-safe: uses internal synchronization to prevent redundant refresh attempts.
+
 	Connection::Ptr getPrimaryConnection();
 		/// Returns a connection to the primary server.
 		/// Returns null if no primary is available.
@@ -202,7 +219,7 @@ class MongoDB_API ReplicaSet
 		/// Returns null if no secondary is available.
 
 	[[nodiscard]] Config configuration() const;
-		// Returns a copy of replica set configuration.
+		/// Returns a copy of replica set configuration.
 
 	[[nodiscard]] TopologyDescription topology() const;
 		/// Returns a copy of the current topology description.
@@ -243,13 +260,22 @@ class MongoDB_API ReplicaSet
 	Connection::Ptr selectServer(const ReadPreference& readPref);
 		/// Selects a server based on read preference and creates a connection.
 
+	Connection::Ptr selectServer(const ReadPreference& readPref,
+		const Poco::Timespan& connectTimeout, const Poco::Timespan& socketTimeout);
+		/// Selects a server based on read preference and creates a connection
+		/// using the specified timeouts.
+
 	Connection::Ptr createConnection(const Net::SocketAddress& address);
 		/// Creates a new connection to the specified address.
 
-	void updateTopologyFromHello(const Net::SocketAddress& address) noexcept;
+	Connection::Ptr createConnection(const Net::SocketAddress& address,
+		const Poco::Timespan& connectTimeout, const Poco::Timespan& socketTimeout);
+		/// Creates a new connection to the specified address using the given timeouts.
+
+	void updateTopologyFromHello(const Net::SocketAddress& address);
 		/// Queries a server with 'hello' command and updates topology.
 
-	void updateTopologyFromAllServers() noexcept;
+	void updateTopologyFromAllServers();
 		/// Queries all known servers and updates topology.
 
 	void parseURI(const std::string& uri);
@@ -264,9 +290,13 @@ class MongoDB_API ReplicaSet
 	std::thread _monitorThread;
 	std::atomic<bool> _stopMonitoring{false};
 	std::atomic<bool> _monitoringActive{false};
+	std::mutex _monitorMutex;
+	std::condition_variable _monitorCV;
+
+	Poco::Random _random;
 
 	std::mutex _serverAvailabilityRetryMutex;
-	std::chrono::steady_clock::time_point _topologyRefreshTime;
+	std::atomic<std::chrono::steady_clock::time_point> _topologyRefreshTime;
 };