Remove the ref count protections

Author: michael stack

fdbrpc/FlowTransport.actor.cpp

@@ -1851,24 +1851,8 @@ void FlowTransport::removePeerReference(const Endpoint& endpoint, bool isStream)
 		return;
 	Reference<Peer> peer = self->getPeer(endpoint.getPrimaryAddress());
 	if (peer) {
-		// Only decrement if the peer has active references (peerReferences > 0)
-		// This prevents race conditions and over-decrementing
-		if (peer->peerReferences > 0) {
-			peer->peerReferences--;
-		} else if (peer->peerReferences == 0) {
-			// Peer already has no references - this can happen in race conditions
-			// Don't decrement further as it would make the count invalid
-			TraceEvent(SevInfo, "PeerReferenceAlreadyZero")
-			    .detail("Address", endpoint.getPrimaryAddress())
-			    .detail("Token", endpoint.token);
-		} else if (peer->peerReferences == -1) {
-			// Peer was created but never had addPeerReference called
-			// This is expected in some race conditions, so don't decrement further
-			TraceEvent(SevInfo, "PeerReferenceNotInitialized")
-			    .detail("Address", endpoint.getPrimaryAddress())
-			    .detail("Token", endpoint.token);
-		} else {
-			// This should not happen with the fixes above, but keep the error for debugging
+		peer->peerReferences--;
+		if (peer->peerReferences < -1) {
 			TraceEvent(SevError, "InvalidPeerReferences")
 			    .detail("References", peer->peerReferences)
 			    .detail("Address", endpoint.getPrimaryAddress())

fdbrpc/include/fdbrpc/fdbrpc.h

@@ -39,21 +39,18 @@ class FlowReceiver : public NetworkMessageReceiver, public NonCopyable {
 	Endpoint endpoint;
 	bool m_isLocalEndpoint;
 	bool m_stream;
-	bool m_peerReferenceAdded;  // Track if we've added a peer reference
-
 protected:
-	FlowReceiver() : m_isLocalEndpoint(false), m_stream(false), m_peerReferenceAdded(false) {}
+	FlowReceiver() : m_isLocalEndpoint(false), m_stream(false) {}
 
 	FlowReceiver(Endpoint const& remoteEndpoint, bool stream)
-	  : endpoint(remoteEndpoint), m_isLocalEndpoint(false), m_stream(stream), m_peerReferenceAdded(false) {
+	  : endpoint(remoteEndpoint), m_isLocalEndpoint(false), m_stream(stream) {
 		FlowTransport::transport().addPeerReference(endpoint, m_stream);
-		m_peerReferenceAdded = true;
 	}
 
 	~FlowReceiver() {
 		if (m_isLocalEndpoint) {
 			FlowTransport::transport().removeEndpoint(endpoint, this);
-		} else if (m_peerReferenceAdded) { // Only remove if a reference was added
+		} else {
 			FlowTransport::transport().removePeerReference(endpoint, m_stream);
 		}
 	}
@@ -65,11 +62,9 @@ class FlowReceiver : public NetworkMessageReceiver, public NonCopyable {
 	void setRemoteEndpoint(Endpoint const& remoteEndpoint, bool stream) {
 		ASSERT(!m_isLocalEndpoint);
 		ASSERT(!endpoint.isValid());
-		ASSERT(!m_peerReferenceAdded);  // Should not already have a reference
 		endpoint = remoteEndpoint;
 		m_stream = stream;
 		FlowTransport::transport().addPeerReference(endpoint, m_stream);
-		m_peerReferenceAdded = true;
 	}
 
 	// If already a remote endpoint, returns that.  Otherwise makes this

fdbserver/MockS3Server.actor.cpp

@@ -997,15 +997,15 @@ ACTOR Future<Void> startMockS3Server(NetworkAddress listenAddress) {
 // Unit Tests for MockS3Server
 TEST_CASE("/fdbserver/MockS3Server/parseS3Request/ValidBucketParameter") {
 	MockS3ServerImpl server;
-	std::string resource = "/test?bucket=mybucket&region=us-east-1";
+	std::string resource = "/test?bucket=testbucket&region=us-east-1";
 	std::string bucket, object;
 	std::map<std::string, std::string> queryParams;
 
 	server.parseS3Request(resource, bucket, object, queryParams);
 
-	ASSERT(bucket == "mybucket");
+	ASSERT(bucket == "testbucket");
 	ASSERT(object == "");
-	ASSERT(queryParams["bucket"] == "mybucket");
+	ASSERT(queryParams["bucket"] == "testbucket");
 	ASSERT(queryParams["region"] == "us-east-1");
 
 	return Void();
@@ -1093,15 +1093,15 @@ TEST_CASE("/fdbserver/MockS3Server/parseS3Request/URLEncodedParameters") {
 
 TEST_CASE("/fdbserver/MockS3Server/parseS3Request/EmptyPath") {
 	MockS3ServerImpl server;
-	std::string resource = "/?bucket=mybucket&region=us-east-1";
+	std::string resource = "/?bucket=testbucket&region=us-east-1";
 	std::string bucket, object;
 	std::map<std::string, std::string> queryParams;
 
 	server.parseS3Request(resource, bucket, object, queryParams);
 
-	ASSERT(bucket == "mybucket");
+	ASSERT(bucket == "testbucket");
 	ASSERT(object == "");
-	ASSERT(queryParams["bucket"] == "mybucket");
+	ASSERT(queryParams["bucket"] == "testbucket");
 	ASSERT(queryParams["region"] == "us-east-1");
 
 	return Void();
@@ -1125,14 +1125,14 @@ TEST_CASE("/fdbserver/MockS3Server/parseS3Request/OnlyBucketInPath") {
 
 TEST_CASE("/fdbserver/MockS3Server/parseS3Request/MultipleParameters") {
 	MockS3ServerImpl server;
-	std::string resource = "/test?bucket=mybucket&region=us-east-1&version=1&encoding=utf8";
+	std::string resource = "/test?bucket=testbucket&region=us-east-1&version=1&encoding=utf8";
 	std::string bucket, object;
 	std::map<std::string, std::string> queryParams;
 
 	server.parseS3Request(resource, bucket, object, queryParams);
 
-	ASSERT(bucket == "mybucket");
-	ASSERT(queryParams["bucket"] == "mybucket");
+	ASSERT(bucket == "testbucket");
+	ASSERT(queryParams["bucket"] == "testbucket");
 	ASSERT(queryParams["region"] == "us-east-1");
 	ASSERT(queryParams["version"] == "1");
 	ASSERT(queryParams["encoding"] == "utf8");
@@ -1143,14 +1143,14 @@ TEST_CASE("/fdbserver/MockS3Server/parseS3Request/MultipleParameters") {
 
 TEST_CASE("/fdbserver/MockS3Server/parseS3Request/ParametersWithoutValues") {
 	MockS3ServerImpl server;
-	std::string resource = "/test?bucket=mybucket&flag&region=us-east-1";
+	std::string resource = "/test?bucket=testbucket&flag&region=us-east-1";
 	std::string bucket, object;
 	std::map<std::string, std::string> queryParams;
 
 	server.parseS3Request(resource, bucket, object, queryParams);
 
-	ASSERT(bucket == "mybucket");
-	ASSERT(queryParams["bucket"] == "mybucket");
+	ASSERT(bucket == "testbucket");
+	ASSERT(queryParams["bucket"] == "testbucket");
 	ASSERT(queryParams["flag"] == ""); // Parameter without value should be empty string
 	ASSERT(queryParams["region"] == "us-east-1");
 

fdbserver/workloads/BackupCorrectness.actor.cpp

@@ -94,7 +94,7 @@ struct BackupAndRestoreCorrectnessWorkload : TestWorkload {
 		shareLogRange = getOption(options, "shareLogRange"_sr, false);
 		defaultBackup = getOption(options, "defaultBackup"_sr, false);
 		backupURL = getOption(options, "backupURL"_sr, "file://simfdb/backups/"_sr).toString();
-		skipDirtyRestore = getOption(options, "skipDirtyRestore"_sr, true);
+		skipDirtyRestore = getOption(options, "skipDirtyRestore"_sr, false);
 		initSnapshotInterval = getOption(options, "initSnapshotInterval"_sr, 0);
 		snapshotInterval = getOption(options, "snapshotInterval"_sr, 30);
 
@@ -610,6 +610,8 @@ struct BackupAndRestoreCorrectnessWorkload : TestWorkload {
 		}
 
 		try {
+			state Future<Void> startRestore = delay(self->restoreAfter);
+
 			// backup
 			wait(delay(self->backupAfter));
 
@@ -653,11 +655,8 @@ struct BackupAndRestoreCorrectnessWorkload : TestWorkload {
 			UidAndAbortedFlagT uidFlag = wait(keyBackedTag.getOrThrow(cx.getReference()));
 			state UID logUid = uidFlag.first;
 			state Key destUidValue = wait(BackupConfig(logUid).destUidValue().getD(cx.getReference()));
-			// Ensure backup reached a restorable/completed state and fetch its container
-			state Reference<IBackupContainer> lastBackupContainer;
-			state UID lastBackupUID;
-			state EBackupState waitState = wait(backupAgent.waitBackup(
-			    cx, self->backupTag.toString(), StopWhenDone::True, &lastBackupContainer, &lastBackupUID));
+			state Reference<IBackupContainer> lastBackupContainer =
+			    wait(BackupConfig(logUid).backupContainer().getD(cx.getReference()));
 
 			// Occasionally start yet another backup that might still be running when we restore
 			if (!self->locked && BUGGIFY) {
@@ -681,11 +680,8 @@ struct BackupAndRestoreCorrectnessWorkload : TestWorkload {
 				}
 			}
 
-			// Wait for backup to complete, then add a small delay before restore
-			TraceEvent("BARW_WaitingForBackupBeforeRestore", randomID).detail("BackupTag", printable(self->backupTag));
-
-			// Add a small delay after backup completion to ensure all metadata is written
-			wait(delay(5.0));
+			CODE_PROBE(!startRestore.isReady(), "Restore starts at specified time");
+			wait(startRestore);
 
 			if (lastBackupContainer && self->performRestore) {
 				if (!self->skipDirtyRestore && deterministicRandom()->random01() < 0.5) {
@@ -706,36 +702,25 @@ struct BackupAndRestoreCorrectnessWorkload : TestWorkload {
 				    .detail("BackupTag", printable(self->backupTag));
 				TraceEvent("BARW_RestoreBegin", randomID).detail("BackupTag", printable(self->backupTag));
 
-				state Reference<IBackupContainer> restoreContainer =
-				    IBackupContainer::openContainer(lastBackupContainer->getURL(),
-				                                    lastBackupContainer->getProxy(),
-				                                    lastBackupContainer->getEncryptionKeyFileName());
-				state BackupDescription restoreDesc;
-				state BackupDescription _initialRestoreDesc = wait(restoreContainer->describeBackup());
-				restoreDesc = _initialRestoreDesc;
-				// Wait until backup becomes restorable to avoid restore_invalid_version
-				state int attempts = 0;
-				loop {
-					if (restoreDesc.maxRestorableVersion.present() || attempts >= 10000)
-						break;
-					wait(delay(1.0));
-					state BackupDescription _loopRestoreDesc = wait(restoreContainer->describeBackup());
-					restoreDesc = _loopRestoreDesc;
-					attempts++;
-				}
-				if (!restoreDesc.maxRestorableVersion.present()) {
-					TraceEvent("BARW_SkipRestoreNotRestorable").detail("BackupTag", printable(self->backupTag));
-					return Void();
-				}
-
-				// Double-check to prevent race condition
-				if (!restoreDesc.maxRestorableVersion.present()) {
-					TraceEvent("BARW_SkipRestoreNotRestorableRace").detail("BackupTag", printable(self->backupTag));
-					return Void();
+				auto container = IBackupContainer::openContainer(lastBackupContainer->getURL(),
+				                                                 lastBackupContainer->getProxy(),
+				                                                 lastBackupContainer->getEncryptionKeyFileName());
+				BackupDescription desc = wait(container->describeBackup());
+
+				state Version targetVersion = -1;
+				if (desc.maxRestorableVersion.present()) {
+					if (deterministicRandom()->random01() < 0.1) {
+						targetVersion = desc.minRestorableVersion.get();
+					} else if (deterministicRandom()->random01() < 0.1) {
+						targetVersion = desc.maxRestorableVersion.get();
+					} else if (deterministicRandom()->random01() < 0.5) {
+						targetVersion = (desc.minRestorableVersion.get() != desc.maxRestorableVersion.get())
+						                    ? deterministicRandom()->randomInt64(desc.minRestorableVersion.get(),
+						                                                         desc.maxRestorableVersion.get())
+						                    : desc.maxRestorableVersion.get();
+					}
 				}
 
-				state Version targetVersion = restoreDesc.maxRestorableVersion.get();
-
 				TraceEvent("BARW_RestoreDebug").detail("TargetVersion", targetVersion);
 
 				state std::vector<Future<Version>> restores;
@@ -895,13 +880,6 @@ struct BackupAndRestoreCorrectnessWorkload : TestWorkload {
 				for (auto& restore : restores) {
 					ASSERT(!restore.isError());
 				}
-
-				// Re-describe the same container instance to print final description
-				{
-					state BackupDescription finalDesc = wait(restoreContainer->describeBackup());
-					wait(finalDesc.resolveVersionTimes(cx));
-					printf("BackupDescription:\n%s\n", finalDesc.toString().c_str());
-				}
 			}
 
 			if (extraBackup.isValid()) {