[DO NOT MERGE] reproduce backup error

fdbclient/S3BlobStore.actor.cpp

@@ -22,7 +22,9 @@
 
 #include "fdbclient/ClientKnobs.h"
 #include "fdbclient/Knobs.h"
+#include "flow/Error.h"
 #include "flow/IConnection.h"
+#include "flow/IRandom.h"
 #include "flow/Trace.h"
 #include "flow/flow.h"
 #include "flow/genericactors.actor.h"
@@ -1005,8 +1007,9 @@ ACTOR Future<Reference<HTTP::IncomingResponse>> doRequest_impl(Reference<S3BlobS
 	state int maxTries = std::min(bstore->knobs.request_tries, bstore->knobs.connect_tries);
 	state int thisTry = 1;
 	state int badRequestCode = 400;
-	state bool s3TokenError = false;
+	state bool s3TokenError = true;
 	state double nextRetryDelay = 2.0;
+	state bool hasDoneError = false;
 
 	loop {
 		state Optional<Error> err;
@@ -1083,8 +1086,13 @@ ACTOR Future<Reference<HTTP::IncomingResponse>> doRequest_impl(Reference<S3BlobS
 					    .detail("V4", CLIENT_KNOBS->HTTP_REQUEST_AWS_V4_HEADER)
 					    .log();
 					wait(bstore->requestRate->getAllowance(1));
-					Future<Reference<HTTP::IncomingResponse>> dryrunResponse = HTTP::doRequest(
-					    rconn.conn, dryrunRequest, bstore->sendRate, &bstore->s_stats.bytes_sent, bstore->recvRate);
+					Future<Reference<HTTP::IncomingResponse>> dryrunResponse =
+					    HTTP::doRequest(rconn.conn,
+					                    dryrunRequest,
+					                    bstore->sendRate,
+					                    &bstore->s_stats.bytes_sent,
+					                    bstore->recvRate,
+					                    &hasDoneError);
 					Reference<HTTP::IncomingResponse> _dryrunR = wait(timeoutError(dryrunResponse, requestTimeout));
 					dryrunR = _dryrunR;
 					std::string s3Error = parseErrorCodeFromS3(dryrunR->data.content);
@@ -1114,7 +1122,8 @@ ACTOR Future<Reference<HTTP::IncomingResponse>> doRequest_impl(Reference<S3BlobS
 				}
 			} catch (Error& e) {
 				// retry with GET failed, but continue to do original request anyway
-				TraceEvent(SevError, "ErrorDuringRetryS3TokenIssue").errorUnsuppressed(e);
+				TraceEvent(SevWarn, "ErrorDuringRetryS3TokenIssue").errorUnsuppressed(e);
+				// continue;
 			}
 			setHeaders(bstore, req);
 			req->resource = getCanonicalURI(bstore, req);
@@ -1153,7 +1162,7 @@ ACTOR Future<Reference<HTTP::IncomingResponse>> doRequest_impl(Reference<S3BlobS
 			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;
+			err = e; // ROOT CAUSE: If there is any error happens to HTTP::doRequest, we do not return the conn.
 		}
 
 		double end = g_network->timer();

fdbrpc/HTTP.actor.cpp

@@ -569,7 +569,8 @@ ACTOR Future<Reference<HTTP::IncomingResponse>> doRequestActor(Reference<IConnec
                                                                Reference<OutgoingRequest> request,
                                                                Reference<IRateControl> sendRate,
                                                                int64_t* pSent,
-                                                               Reference<IRateControl> recvRate) {
+                                                               Reference<IRateControl> recvRate,
+                                                               bool* hasDoneError = nullptr) {
 	state TraceEvent event(SevDebug, "HTTPRequest");
 
 	// There is no standard http request id header field, so either a global default can be set via a knob
@@ -650,7 +651,7 @@ ACTOR Future<Reference<HTTP::IncomingResponse>> doRequestActor(Reference<IConnec
 				trySend = deterministicRandom()->randomInt(1, 10);
 			}
 			wait(sendRate->getAllowance(trySend));
-			int len = conn->write(request->data.content->getUnsent(), trySend);
+			int len = conn->write(request->data.content->getUnsent(), trySend, hasDoneError);
 			if (pSent != nullptr)
 				*pSent += len;
 			sendRate->returnUnused(trySend - len);
@@ -757,8 +758,9 @@ Future<Reference<IncomingResponse>> doRequest(Reference<IConnection> conn,
                                               Reference<OutgoingRequest> request,
                                               Reference<IRateControl> sendRate,
                                               int64_t* pSent,
-                                              Reference<IRateControl> recvRate) {
-	return doRequestActor(conn, request, sendRate, pSent, recvRate);
+                                              Reference<IRateControl> recvRate,
+                                              bool* hasDoneError) {
+	return doRequestActor(conn, request, sendRate, pSent, recvRate, hasDoneError);
 }
 
 ACTOR Future<Void> sendProxyConnectRequest(Reference<IConnection> conn,

fdbrpc/HTTPServer.actor.cpp

@@ -335,7 +335,7 @@ ACTOR Future<Reference<HTTP::IncomingResponse>> doHelloWorldReq(Reference<IConne
 	pw.serializeBytes(hello);
 
 	Reference<HTTP::IncomingResponse> response =
-	    wait(timeoutError(HTTP::doRequest(conn, req, sendReceiveRate, &bytes_sent, sendReceiveRate), 30.0));
+	    wait(timeoutError(HTTP::doRequest(conn, req, sendReceiveRate, &bytes_sent, sendReceiveRate, nullptr), 30.0));
 
 	std::string expectedContent = "Hello World Response!";
 
@@ -367,7 +367,7 @@ ACTOR Future<Reference<HTTP::IncomingResponse>> doHelloWorldErrorReq(Reference<I
 	req->data.contentLen = 0;
 
 	Reference<HTTP::IncomingResponse> response =
-	    wait(timeoutError(HTTP::doRequest(conn, req, sendReceiveRate, &bytes_sent, sendReceiveRate), 30.0));
+	    wait(timeoutError(HTTP::doRequest(conn, req, sendReceiveRate, &bytes_sent, sendReceiveRate, nullptr), 30.0));
 
 	ASSERT(response->code == 500);
 
@@ -393,7 +393,7 @@ ACTOR Future<Reference<HTTP::IncomingResponse>> doHelloBadMD5Req(Reference<IConn
 	pw.serializeBytes(hello);
 
 	Reference<HTTP::IncomingResponse> response =
-	    wait(timeoutError(HTTP::doRequest(conn, req, sendReceiveRate, &bytes_sent, sendReceiveRate), 30.0));
+	    wait(timeoutError(HTTP::doRequest(conn, req, sendReceiveRate, &bytes_sent, sendReceiveRate, nullptr), 30.0));
 
 	// should have gotten error
 	ASSERT(false);

fdbrpc/SimExternalConnection.actor.cpp

@@ -100,7 +100,7 @@ int SimExternalConnection::read(uint8_t* begin, uint8_t* end) {
 	return toRead;
 }
 
-int SimExternalConnection::write(SendBuffer const* buffer, int limit) {
+int SimExternalConnection::write(SendBuffer const* buffer, int limit, bool* hasDoneError) {
 	boost::system::error_code err;
 	bool triggerReaders = (socket.available() == 0);
 	int bytesSent = socket.write_some(

fdbrpc/include/fdbrpc/HTTP.h

@@ -120,7 +120,8 @@ Future<Reference<IncomingResponse>> doRequest(Reference<IConnection> conn,
                                               Reference<OutgoingRequest> request,
                                               Reference<IRateControl> sendRate,
                                               int64_t* pSent,
-                                              Reference<IRateControl> recvRate);
+                                              Reference<IRateControl> recvRate,
+                                              bool* hasDoneError = nullptr);
 
 // Connect to proxy, send CONNECT command, and connect to the remote host.
 Future<Reference<IConnection>> proxyConnect(const std::string& remoteHost,

fdbrpc/include/fdbrpc/SimExternalConnection.h

@@ -46,7 +46,7 @@ class SimExternalConnection final : public IConnection, public ReferenceCounted<
 	Future<Void> onWritable() override;
 	Future<Void> onReadable() override;
 	int read(uint8_t* begin, uint8_t* end) override;
-	int write(SendBuffer const* buffer, int limit) override;
+	int write(SendBuffer const* buffer, int limit, bool* hasDoneError = nullptr) override;
 	NetworkAddress getPeerAddress() const override;
 	bool hasTrustedPeer() const override;
 	UID getDebugID() const override;

fdbrpc/sim2.actor.cpp

@@ -418,7 +418,7 @@ struct Sim2Conn final : IConnection, ReferenceCounted<Sim2Conn> {
 
 	// Writes as many bytes as possible from the given SendBuffer chain into the write buffer and returns the number of
 	// bytes written (might be 0) (or may throw an error if the connection dies)
-	int write(SendBuffer const* buffer, int limit) override {
+	int write(SendBuffer const* buffer, int limit, bool* hasDoneError = nullptr) override {
 		rollRandomClose();
 		ASSERT(limit > 0);
 

flow/Net2.actor.cpp

@@ -489,13 +489,18 @@ class Connection final : public IConnection, ReferenceCounted<Connection> {
 
 	// Writes as many bytes as possible from the given SendBuffer chain into the write buffer and returns the number of
 	// bytes written (might be 0)
-	int write(SendBuffer const* data, int limit) override {
+	int write(SendBuffer const* data, int limit, bool* hasDoneError = nullptr) override {
 		boost::system::error_code err;
 		++g_net2->countWrites;
 
 		size_t sent = socket.write_some(
 		    boost::iterator_range<SendBufferIterator>(SendBufferIterator(data, limit), SendBufferIterator()), err);
 
+		if (hasDoneError != nullptr && !(*hasDoneError)) {
+			*hasDoneError = true;
+			err = boost::system::error_code(boost::system::errc::operation_canceled, boost::system::generic_category());
+		}
+
 		if (err) {
 			// Since there was an error, sent's value can't be used to infer that the buffer has data and the limit is
 			// positive so check explicitly.
@@ -1124,7 +1129,7 @@ class SSLConnection final : public IConnection, ReferenceCounted<SSLConnection>
 
 	// Writes as many bytes as possible from the given SendBuffer chain into the write buffer and returns the number of
 	// bytes written (might be 0)
-	int write(SendBuffer const* data, int limit) override {
+	int write(SendBuffer const* data, int limit, bool* hasDoneError = nullptr) override {
 #ifdef __APPLE__
 		// For some reason, writing ssl_sock with more than 2016 bytes when socket is writeable sometimes results in a
 		// broken pipe error.
@@ -1136,6 +1141,11 @@ class SSLConnection final : public IConnection, ReferenceCounted<SSLConnection>
 		size_t sent = ssl_sock.write_some(
 		    boost::iterator_range<SendBufferIterator>(SendBufferIterator(data, limit), SendBufferIterator()), err);
 
+		if (hasDoneError != nullptr && !(*hasDoneError)) {
+			*hasDoneError = true;
+			err = boost::system::error_code(boost::system::errc::operation_canceled, boost::system::generic_category());
+		}
+
 		if (err) {
 			// Since there was an error, sent's value can't be used to infer that the buffer has data and the limit is
 			// positive so check explicitly.
@@ -2296,7 +2306,7 @@ TEST_CASE("noSim/flow/Net2/onMainThreadFIFO") {
 	return Void();
 }
 
-void net2_test(){
+void net2_test() {
 	/*
 	g_network = newNet2();  // for promise serialization below
 

flow/include/flow/IConnection.h

@@ -71,7 +71,9 @@ class IConnection {
 	// entire byte set that the caller was attempting to write even if it is unable to write all of it immediately. Due
 	// to limitations of TLSConnection, callers must also avoid reallocations that reduce the amount of written data in
 	// the first buffer in the chain.
-	virtual int write(SendBuffer const* buffer, int limit = std::numeric_limits<int>::max()) = 0;
+	virtual int write(SendBuffer const* buffer,
+	                  int limit = std::numeric_limits<int>::max(),
+	                  bool* hasDoneError = nullptr) = 0;
 
 	// Returns the network address and port of the other end of the connection.  In the case of an incoming connection,
 	// this may not be an address we can connect to!