From 9ec9f41d34da401ba7d7a81bc87710004a401ac6 Mon Sep 17 00:00:00 2001
From: Evan Tschannen <ejt@apple.com>
Date: Tue, 1 Oct 2019 18:52:07 -0700
Subject: [PATCH] backup and DR would not share mutations if they were started
 on different versions of FDB

---
 fdbclient/DatabaseBackupAgent.actor.cpp | 45 +++++++++++++++----------
 fdbclient/FileBackupAgent.actor.cpp     | 17 ++++++----
 flow/ProtocolVersion.h                  |  1 +
 3 files changed, 40 insertions(+), 23 deletions(-)

diff --git a/fdbclient/DatabaseBackupAgent.actor.cpp b/fdbclient/DatabaseBackupAgent.actor.cpp
index 83bf983c10..ee350f1e0a 100644
--- a/fdbclient/DatabaseBackupAgent.actor.cpp
+++ b/fdbclient/DatabaseBackupAgent.actor.cpp
@@ -1296,19 +1296,25 @@ namespace dbBackup {
 					}
 
 					if (backupRanges.size() == 1) {
-						state Key destUidLookupPath = BinaryWriter::toValue(backupRanges[0], IncludeVersion()).withPrefix(destUidLookupPrefix);
-						Optional<Key> existingDestUidValue = wait(srcTr->get(destUidLookupPath));
-						if (existingDestUidValue.present()) {
-							if (destUidValue == existingDestUidValue.get()) {
-								// due to unknown commit result
-								break;
-							} else {
-								// existing backup/DR is running
-								return Void();
+						Standalone<RangeResultRef> existingDestUidValues = wait(srcTr->getRange(KeyRangeRef(destUidLookupPrefix, strinc(destUidLookupPrefix)), CLIENT_KNOBS->TOO_MANY));
+						bool found = false;
+						for(auto it : existingDestUidValues) {
+							if( BinaryReader::fromStringRef<KeyRangeRef>(it.key.removePrefix(destUidLookupPrefix), IncludeVersion()) == backupRanges[0] ) {
+								if(destUidValue != it.value) {
+									// existing backup/DR is running
+									return Void();
+								} else {
+									// due to unknown commit result
+									found = true;
+									break;
+								}
 							}
 						}
-						
-						srcTr->set(destUidLookupPath, destUidValue);
+						if(found) {
+							break;
+						}
+
+						srcTr->set(BinaryWriter::toValue(backupRanges[0], IncludeVersion(ProtocolVersion::withSharedMutations())).withPrefix(destUidLookupPrefix), destUidValue);
 					}
 
 					Key versionKey = logUidValue.withPrefix(destUidValue).withPrefix(backupLatestVersionsPrefix);
@@ -1466,13 +1472,18 @@ namespace dbBackup {
 
 					// Initialize destUid
 					if (backupRanges.size() == 1) {
-						state Key destUidLookupPath = BinaryWriter::toValue(backupRanges[0], IncludeVersion()).withPrefix(destUidLookupPrefix);
-						Optional<Key> existingDestUidValue = wait(srcTr->get(destUidLookupPath));
-						if (existingDestUidValue.present()) {
-							destUidValue = existingDestUidValue.get();
-						} else {
+						Standalone<RangeResultRef> existingDestUidValues = wait(srcTr->getRange(KeyRangeRef(destUidLookupPrefix, strinc(destUidLookupPrefix)), CLIENT_KNOBS->TOO_MANY));
+						bool found = false;
+						for(auto it : existingDestUidValues) {
+							if( BinaryReader::fromStringRef<KeyRangeRef>(it.key.removePrefix(destUidLookupPrefix), IncludeVersion()) == backupRanges[0] ) {
+								destUidValue = it.value;
+								found = true;
+								break;
+							}
+						}
+						if( !found ) {
 							destUidValue = BinaryWriter::toValue(deterministicRandom()->randomUniqueID(), Unversioned());
-							srcTr->set(destUidLookupPath, destUidValue);
+							srcTr->set(BinaryWriter::toValue(backupRanges[0], IncludeVersion(ProtocolVersion::withSharedMutations())).withPrefix(destUidLookupPrefix), destUidValue);
 						}
 					}
 
diff --git a/fdbclient/FileBackupAgent.actor.cpp b/fdbclient/FileBackupAgent.actor.cpp
index 6c245f8230..bf6fceab70 100644
--- a/fdbclient/FileBackupAgent.actor.cpp
+++ b/fdbclient/FileBackupAgent.actor.cpp
@@ -3619,13 +3619,18 @@ public:
 
 		state Key destUidValue(BinaryWriter::toValue(uid, Unversioned()));
 		if (normalizedRanges.size() == 1) {
-			state Key destUidLookupPath = BinaryWriter::toValue(normalizedRanges[0], IncludeVersion()).withPrefix(destUidLookupPrefix);
-			Optional<Key> existingDestUidValue = wait(tr->get(destUidLookupPath));
-			if (existingDestUidValue.present()) {
-				destUidValue = existingDestUidValue.get();
-			} else {
+			Standalone<RangeResultRef> existingDestUidValues = wait(tr->getRange(KeyRangeRef(destUidLookupPrefix, strinc(destUidLookupPrefix)), CLIENT_KNOBS->TOO_MANY));
+			bool found = false;
+			for(auto it : existingDestUidValues) {
+				if( BinaryReader::fromStringRef<KeyRangeRef>(it.key.removePrefix(destUidLookupPrefix), IncludeVersion()) == normalizedRanges[0] ) {
+					destUidValue = it.value;
+					found = true;
+					break;
+				}
+			}
+			if( !found ) {
 				destUidValue = BinaryWriter::toValue(deterministicRandom()->randomUniqueID(), Unversioned());
-				tr->set(destUidLookupPath, destUidValue);
+				tr->set(BinaryWriter::toValue(normalizedRanges[0], IncludeVersion(ProtocolVersion::withSharedMutations())).withPrefix(destUidLookupPrefix), destUidValue);
 			}
 		}
 
diff --git a/flow/ProtocolVersion.h b/flow/ProtocolVersion.h
index ed82ae792f..00f7b2108e 100644
--- a/flow/ProtocolVersion.h
+++ b/flow/ProtocolVersion.h
@@ -79,6 +79,7 @@ public: // introduced features
 	PROTOCOL_VERSION_FEATURE(0x0FDB00A400040000LL, OpenDatabase);
 	PROTOCOL_VERSION_FEATURE(0x0FDB00A446020000LL, Locality);
 	PROTOCOL_VERSION_FEATURE(0x0FDB00A460010000LL, MultiGenerationTLog);
+	PROTOCOL_VERSION_FEATURE(0x0FDB00A460010000LL, SharedMutations);
 	PROTOCOL_VERSION_FEATURE(0x0FDB00A551000000LL, MultiVersionClient);
 	PROTOCOL_VERSION_FEATURE(0x0FDB00A560010000LL, TagLocality);
 	PROTOCOL_VERSION_FEATURE(0x0FDB00B060000000LL, Fearless);