Fix S3/REST infrastructure: connection lifecycle and API compliance (#12447)

* Fix S3/REST infrastructure: connection lifecycle and API compliance
(First of three PR to add backup and restore via s3/mocks3).

Fix issues in S3 and REST client infrastructure that were causing
simulation failures and API compliance problems.

PART 1: Sim2Conn Connection Lifecycle Fixes

Problem: Connection pools were destroying connections without explicitly closing
them, violating Sim2Conn's assertion (!opened || closedByCaller) in simulation.

Fixes:
1. Add destructors to connection pool classes that close all pooled connections
   before destruction (RESTConnectionPool, S3BlobStoreEndpoint::ConnectionPoolData)

2. Close connections on error paths in doRequest_impl() before they go out of
   scope (both REST and S3)

3. Replace static globalConnectionPool with getGlobalConnectionPool() returning
   process-local pools in simulation

4. Add ReusableConnection copy control to detect cross-process copies

PART 2: S3 API Compliance and Error Handling

Problem: S3 API spec violations and incorrect error handling causing backup failures.

Fixes:
1. Strip leading slashes from S3 object keys (S3 spec: keys shouldn't start with '/')
   - constructResourcePath() now removes all leading '/' characters

2. URL decode object names from S3 XML responses (S3 returns "test%20file.txt")
   - listObjectsStream_impl(), BackupContainerS3BlobStore list functions

3. Fix guessRegionFromDomain() crash with localhost/127.0.0.1 (used by MockS3)

4. Only parse S3 error codes for 4xx/5xx responses, not 2xx success responses

5. Add connection validation ASSERT before handshake

6. Use flat listing in BackupContainerS3BlobStore to find all nested files

- Fixes all Sim2Conn assertion failures in S3/REST tests
- Enables S3 backups to work correctly with proper object key handling
- Fixes MockS3 testing with localhost addresses
- Simulation-only changes (g_network->isSimulated()) have no production impact
- S3 API compliance changes apply to both simulation and production

* Fix compile issue

* Formatting

* Fix rconn scope issue in S3BlobStore error handler

* Fix s3client_test to use full object path in list verification

The test was checking for 'ls_test/sub1/file2_1' but the actual S3 object
key is 's3client/ls_test/sub1/file2_1' (the full path including the path_prefix).

S3 list operations return the complete object key, not a path relative to
the prefix. This broke after the object key normalization changes that strip
leading slashes, which changed how paths were being constructed.

Fix: Extract the full path prefix from the URL and use it when verifying
the listing output, so we check for the correct full object paths.

* Formatting

* Address review improvement suggestion

Author: saintstack

fdbclient/BackupContainerS3BlobStore.actor.cpp

@@ -22,6 +22,7 @@
 #include "fdbclient/BackupContainerS3BlobStore.h"
 #include "fdbrpc/AsyncFileEncrypted.h"
 #include "fdbrpc/AsyncFileReadAhead.actor.h"
+#include "fdbrpc/HTTP.h"
 #include "flow/actorcompiler.h" // This must be the last #include.
 
 class BackupContainerS3BlobStoreImpl {
@@ -38,8 +39,10 @@ class BackupContainerS3BlobStoreImpl {
 		S3BlobStoreEndpoint::ListResult contents = wait(bstore->listObjects(bucket, basePath));
 		std::vector<std::string> results;
 		for (const auto& f : contents.objects) {
+			// URL decode the object name since S3 XML responses contain URL-encoded names
+			std::string decodedName = HTTP::urlDecode(f.name);
 			results.push_back(
-			    bstore->getResourceURL(f.name.substr(basePath.size()), format("bucket=%s", bucket.c_str())));
+			    bstore->getResourceURL(decodedName.substr(basePath.size()), format("bucket=%s", bucket.c_str())));
 		}
 		return results;
 	}
@@ -85,12 +88,15 @@ class BackupContainerS3BlobStoreImpl {
 			return pathFilter(folderPath.substr(prefixTrim));
 		};
 
-		state S3BlobStoreEndpoint::ListResult result = wait(bc->m_bstore->listObjects(
-		    bc->m_bucket, bc->dataPath(path), '/', std::numeric_limits<int>::max(), rawPathFilter));
+		// Use flat listing for backup files to ensure all files are found regardless of directory structure
+		state S3BlobStoreEndpoint::ListResult result =
+		    wait(bc->m_bstore->listObjects(bc->m_bucket, bc->dataPath(path), Optional<char>(), 0, rawPathFilter));
 		BackupContainerFileSystem::FilesAndSizesT files;
 		for (const auto& o : result.objects) {
 			ASSERT(o.name.size() >= prefixTrim);
-			files.push_back({ o.name.substr(prefixTrim), o.size });
+			// URL decode the object name since S3 XML responses contain URL-encoded names
+			std::string decodedName = HTTP::urlDecode(o.name);
+			files.push_back({ decodedName.substr(prefixTrim), o.size });
 		}
 		return files;
 	}

fdbclient/RESTClient.actor.cpp

@@ -138,21 +138,22 @@ ACTOR Future<Reference<HTTP::IncomingResponse>> doRequest_impl(Reference<RESTCli
 		state bool connectionEstablished = false;
 
 		state Reference<HTTP::IncomingResponse> r;
+		state RESTConnectionPool::ReusableConnection rconn;
 
 		try {
 			// Start connecting
 			Future<RESTConnectionPool::ReusableConnection> frconn =
 			    client->conectionPool->connect(connectPoolKey, url.connType.secure, client->knobs.max_connection_life);
 
 			// Finish connecting, do request
-			state RESTConnectionPool::ReusableConnection rconn =
-			    wait(timeoutError(frconn, client->knobs.connect_timeout));
+			wait(store(rconn, timeoutError(frconn, client->knobs.connect_timeout)));
 			connectionEstablished = true;
 
 			remoteAddress = rconn.conn->getPeerAddress();
-			Reference<HTTP::IncomingResponse> _r = wait(timeoutError(
-			    HTTP::doRequest(rconn.conn, req, sendReceiveRate, &statsPtr->bytes_sent, sendReceiveRate), reqTimeout));
-			r = _r;
+			wait(store(
+			    r,
+			    timeoutError(HTTP::doRequest(rconn.conn, req, sendReceiveRate, &statsPtr->bytes_sent, sendReceiveRate),
+			                 reqTimeout)));
 
 			// Since the response was parsed successfully (which is why we are here) reuse the connection unless we
 			// received the "Connection: close" header.
@@ -161,6 +162,14 @@ ACTOR Future<Reference<HTTP::IncomingResponse>> doRequest_impl(Reference<RESTCli
 			}
 			rconn.conn.clear();
 		} catch (Error& e) {
+			// Close the connection on error to satisfy Sim2Conn's assertion in simulation.
+			// If rconn holds a connection that hasn't been returned to the pool,
+			// we must close it before it's destroyed (when rconn goes out of scope).
+			// This sets closedByCaller = true in Sim2Conn, preventing assertion failures.
+			if (connectionEstablished && rconn.conn.isValid()) {
+				rconn.conn->close();
+				rconn.conn.clear();
+			}
 			if (e.code() == error_code_actor_cancelled) {
 				throw;
 			}

fdbclient/RESTUtils.actor.cpp

@@ -30,6 +30,24 @@
 
 #include "flow/actorcompiler.h" // always the last include
 
+// RESTConnectionPool destructor implementation
+RESTConnectionPool::~RESTConnectionPool() {
+	// In simulation, explicitly close all pooled connections before destruction.
+	// This satisfies Sim2Conn's assertion: !opened || closedByCaller
+	// Without this, connections would be destroyed without being closed, causing assertion failures.
+	if (g_network && g_network->isSimulated()) {
+		for (auto& kv : connectionPoolMap) {
+			while (!kv.second.empty()) {
+				ReusableConnection& rconn = kv.second.front();
+				if (rconn.conn.isValid()) {
+					rconn.conn->close();
+				}
+				kv.second.pop();
+			}
+		}
+	}
+}
+
 const std::unordered_map<std::string, RESTConnectionType> RESTConnectionType::supportedConnTypes = {
 	{ "http", RESTConnectionType("http", RESTConnectionType::NOT_SECURE_CONNECTION) },
 	{ "https", RESTConnectionType("https", RESTConnectionType::SECURE_CONNECTION) }

fdbclient/S3BlobStore.actor.cpp

@@ -20,13 +20,32 @@
 
 #include "fdbclient/S3BlobStore.h"
 
+#include <sstream>
+#include "fdbrpc/HTTP.h"
 #include "fdbclient/ClientKnobs.h"
 #include "fdbclient/Knobs.h"
 #include "flow/FastRef.h"
 #include "flow/IConnection.h"
 #include "flow/Trace.h"
 #include "flow/flow.h"
 #include "flow/genericactors.actor.h"
+
+// S3BlobStoreEndpoint::ConnectionPoolData destructor implementation
+S3BlobStoreEndpoint::ConnectionPoolData::~ConnectionPoolData() {
+	// In simulation, explicitly close all pooled connections before destruction.
+	// This satisfies Sim2Conn's assertion: !opened || closedByCaller
+	// Without this, connections would be destroyed without being closed, causing assertion failures.
+	if (g_network && g_network->isSimulated()) {
+		while (!pool.empty()) {
+			ReusableConnection& rconn = pool.front();
+			if (rconn.conn.isValid()) {
+				rconn.conn->close();
+			}
+			pool.pop();
+		}
+	}
+}
+
 #include "md5/md5.h"
 #include "libb64/encode.h"
 #include "fdbclient/sha1/SHA1.h"
@@ -76,9 +95,6 @@ S3BlobStoreEndpoint::Stats S3BlobStoreEndpoint::s_stats;
 std::unique_ptr<S3BlobStoreEndpoint::BlobStats> S3BlobStoreEndpoint::blobStats;
 Future<Void> S3BlobStoreEndpoint::statsLogger = Never();
 
-std::unordered_map<BlobStoreConnectionPoolKey, Reference<S3BlobStoreEndpoint::ConnectionPoolData>>
-    S3BlobStoreEndpoint::globalConnectionPool;
-
 S3BlobStoreEndpoint::BlobKnobs::BlobKnobs() {
 	secure_connection = 1;
 	connect_tries = CLIENT_KNOBS->BLOBSTORE_CONNECT_TRIES;
@@ -199,6 +215,11 @@ std::string S3BlobStoreEndpoint::BlobKnobs::getURLParameters() const {
 }
 
 std::string guessRegionFromDomain(std::string domain) {
+	// Special case for localhost/127.0.0.1 to prevent basic_string exception
+	if (domain == "127.0.0.1" || domain == "localhost") {
+		return "us-east-1";
+	}
+
 	static const std::vector<const char*> knownServices = { "s3.", "cos.", "oss-", "obs." };
 	boost::algorithm::to_lower(domain);
 
@@ -446,11 +467,15 @@ std::string S3BlobStoreEndpoint::constructResourcePath(const std::string& bucket
 	}
 
 	if (!object.empty()) {
-		// Don't add a slash if the object starts with one
-		if (!object.starts_with("/")) {
+		// S3 object keys should not start with '/'. Strip any leading slashes.
+		std::string cleanedObject = object;
+		while (!cleanedObject.empty() && cleanedObject[0] == '/') {
+			cleanedObject = cleanedObject.substr(1);
+		}
+		if (!cleanedObject.empty()) {
 			resource += "/";
+			resource += cleanedObject;
 		}
-		resource += object;
 	}
 
 	return resource;
@@ -843,6 +868,10 @@ ACTOR Future<S3BlobStoreEndpoint::ReusableConnection> connect_impl(Reference<S3B
 	} else {
 		wait(store(conn, INetworkConnections::net()->connect(host, service, isTLS)));
 	}
+
+	// Ensure connection is valid before handshake
+	ASSERT(conn.isValid());
+
 	wait(conn->connectHandshake());
 
 	TraceEvent("S3BlobStoreEndpointNewConnectionSuccess")
@@ -1074,6 +1103,7 @@ ACTOR Future<Reference<HTTP::IncomingResponse>> doRequest_impl(Reference<S3BlobS
 		state bool reusingConn = false;
 		state bool fastRetry = false;
 		state bool simulateS3TokenError = false;
+		state S3BlobStoreEndpoint::ReusableConnection rconn; // Moved outside try block for error handler access
 
 		try {
 			// Start connecting
@@ -1095,8 +1125,7 @@ ACTOR Future<Reference<HTTP::IncomingResponse>> doRequest_impl(Reference<S3BlobS
 			}
 
 			// Finish connecting, do request
-			state S3BlobStoreEndpoint::ReusableConnection rconn =
-			    wait(timeoutError(frconn, bstore->knobs.connect_timeout));
+			wait(store(rconn, timeoutError(frconn, bstore->knobs.connect_timeout)));
 			connectionEstablished = true;
 			connID = rconn.conn->getDebugID();
 			reqStartTimer = g_network->timer();
@@ -1140,7 +1169,11 @@ ACTOR Future<Reference<HTTP::IncomingResponse>> doRequest_impl(Reference<S3BlobS
 					    rconn.conn, dryrunRequest, bstore->sendRate, &bstore->s_stats.bytes_sent, bstore->recvRate);
 					Reference<HTTP::IncomingResponse> _dryrunR = wait(timeoutError(dryrunResponse, requestTimeout));
 					dryrunR = _dryrunR;
-					std::string s3Error = parseErrorCodeFromS3(dryrunR->data.content);
+					// Only parse S3 error code for error responses (4xx/5xx), not successful responses (2xx)
+					std::string s3Error;
+					if (dryrunR->code >= 400) {
+						s3Error = parseErrorCodeFromS3(dryrunR->data.content);
+					}
 					if (dryrunR->code == badRequestCode && isS3TokenError(s3Error)) {
 						// authentication fails and s3 token error persists, retry with a HEAD dryrun request
 						// to avoid sending duplicate data indefinitly to save network bandwidth
@@ -1206,9 +1239,16 @@ ACTOR Future<Reference<HTTP::IncomingResponse>> doRequest_impl(Reference<S3BlobS
 			    .errorUnsuppressed(e)
 			    .detail("Verb", verb)
 			    .detail("Resource", resource);
+			// Close the connection on error to satisfy Sim2Conn's assertion in simulation.
+			// If rconn holds a connection that hasn't been returned to the pool,
+			// we must close it before it's destroyed (when rconn goes out of scope).
+			// This sets closedByCaller = true in Sim2Conn, preventing assertion failures.
+			if (connectionEstablished && rconn.conn.isValid()) {
+				rconn.conn->close();
+				rconn.conn.clear();
+			}
 			if (e.code() == error_code_actor_cancelled)
 				throw;
-			// TODO: should this also do rconn.conn.clear()? (would need to extend lifetime outside of try block)
 			err = e;
 		}
 
@@ -1263,7 +1303,12 @@ ACTOR Future<Reference<HTTP::IncomingResponse>> doRequest_impl(Reference<S3BlobS
 
 		if (!err.present()) {
 			event.detail("ResponseCode", r->code);
-			std::string s3Error = parseErrorCodeFromS3(r->data.content);
+			// Only parse S3 error code for real error responses (4xx/5xx), not successful responses (2xx)
+			// Skip parsing for simulated errors where response content is still binary data
+			std::string s3Error;
+			if (r->code >= 400 && !simulateS3TokenError) {
+				s3Error = parseErrorCodeFromS3(r->data.content);
+			}
 			event.detail("S3ErrorCode", s3Error);
 			if (r->code == badRequestCode) {
 				if (isS3TokenError(s3Error) || simulateS3TokenError) {
@@ -1466,7 +1511,8 @@ ACTOR Future<Void> listObjectsStream_impl(Reference<S3BlobStoreEndpoint> bstore,
 					if (key == nullptr) {
 						throw http_bad_response();
 					}
-					object.name = key->value();
+					// URL decode the object name since S3 XML responses contain URL-encoded names
+					object.name = HTTP::urlDecode(key->value());
 
 					xml_node<>* size = n->first_node("Size");
 					if (size == nullptr) {
@@ -2038,15 +2084,12 @@ ACTOR Future<int> readObject_impl(Reference<S3BlobStoreEndpoint> bstore,
 			throw io_error();
 		}
 
-		try {
-			// Copy the output bytes, server could have sent more or less bytes than requested so copy at most length
-			// bytes
-			memcpy(data, r->data.content.data(), std::min<int64_t>(r->data.contentLen, length));
-			return r->data.contentLen;
-		} catch (Error& e) {
-			TraceEvent(SevWarn, "S3BlobStoreReadObjectMemcpyError").detail("Error", e.what());
-			throw io_error();
-		}
+		// Copy the output bytes, server could have sent more or less bytes than requested so copy at most length
+		// bytes
+		int bytesToCopy = std::min<int64_t>(r->data.contentLen, length);
+		memcpy(data, r->data.content.data(), bytesToCopy);
+		// Return the number of bytes actually copied
+		return bytesToCopy;
 	} catch (Error& e) {
 		TraceEvent(SevWarn, "S3BlobStoreEndpoint_ReadError")
 		    .error(e)
@@ -2490,4 +2533,42 @@ TEST_CASE("/backup/s3/virtual_hosting_list_resource_path") {
 	ASSERT(listResource == "/test-bucket"); // Bucket is part of path
 
 	return Void();
-}
+}
+
+TEST_CASE("/backup/s3/constructResourcePath") {
+	// Test that leading slashes in object keys are properly stripped
+	// S3 object keys should not start with '/', but if passed, we normalize them
+
+	std::string url = "blobstore://s3.us-west-2.amazonaws.com?bucket=test-bucket";
+	std::string resource;
+	std::string error;
+	S3BlobStoreEndpoint::ParametersT parameters;
+	Reference<S3BlobStoreEndpoint> s3 = S3BlobStoreEndpoint::fromString(url, {}, &resource, &error, &parameters);
+
+	// Test normal object key (no leading slash)
+	ASSERT(s3->constructResourcePath("test-bucket", "normal/path/file.txt") == "/test-bucket/normal/path/file.txt");
+
+	// Test single leading slash - should be stripped
+	ASSERT(s3->constructResourcePath("test-bucket", "/leading/slash/file.txt") ==
+	       "/test-bucket/leading/slash/file.txt");
+
+	// Test multiple leading slashes - all should be stripped
+	ASSERT(s3->constructResourcePath("test-bucket", "///multiple/slashes.txt") == "/test-bucket/multiple/slashes.txt");
+
+	// Test object key that is only slashes - should result in empty object path
+	ASSERT(s3->constructResourcePath("test-bucket", "///") == "/test-bucket");
+
+	// Test empty object key
+	ASSERT(s3->constructResourcePath("test-bucket", "") == "/test-bucket");
+
+	// Test virtual hosting mode (bucket in hostname)
+	url = "blobstore://test-bucket.s3.us-west-2.amazonaws.com";
+	s3 = S3BlobStoreEndpoint::fromString(url, {}, &resource, &error, &parameters);
+
+	// Virtual hosting mode doesn't include bucket in path
+	ASSERT(s3->constructResourcePath("test-bucket", "normal/file.txt") == "/normal/file.txt");
+	ASSERT(s3->constructResourcePath("test-bucket", "/leading/slash.txt") == "/leading/slash.txt");
+	ASSERT(s3->constructResourcePath("test-bucket", "") == "");
+
+	return Void();
+}

fdbclient/include/fdbclient/RESTUtils.h

@@ -49,6 +49,9 @@ class RESTConnectionPool : public ReferenceCounted<RESTConnectionPool> {
 	struct ReusableConnection {
 		Reference<IConnection> conn;
 		double expirationTime;
+
+		ReusableConnection() : expirationTime(0) {}
+		ReusableConnection(Reference<IConnection> c, double exp) : conn(c), expirationTime(exp) {}
 	};
 
 	// Maximum number of connections cached in the connection-pool.
@@ -58,6 +61,10 @@ class RESTConnectionPool : public ReferenceCounted<RESTConnectionPool> {
 
 	RESTConnectionPool(const int maxConnsPerKey) : maxConnPerConnectKey(maxConnsPerKey) {}
 
+	// Destructor implementation in RESTUtils.actor.cpp
+	// In simulation, explicitly closes all pooled connections before destruction
+	~RESTConnectionPool();
+
 	// Routine is responsible to provide an usable TCP connection object; it reuses an active connection from
 	// connection-pool if available, otherwise, establish a new TCP connection
 	Future<ReusableConnection> connect(RESTConnectionPoolKey connectKey, const bool isSecure, const int maxConnLife);