Use a local keyRangeMap to avoid assigning ranges not part of client ranges.

fdbserver/BlobManager.actor.cpp

@@ -1019,6 +1019,7 @@ ACTOR Future<Void> recoverBlobManager(BlobManagerData* bmData) {
 	//    If the worker already had the range, this is a no-op. If the worker didn't have it, it will
 	//    begin persisting it. The worker that had the same range before will now be at a lower seqno.
 
+	state KeyRangeMap<Optional<UID>> workerAssignments;
 	state Reference<ReadYourWritesTransaction> tr = makeReference<ReadYourWritesTransaction>(bmData->db);
 
 	// Step 1. Get the latest known mapping of granules to blob workers (i.e. assignments)
@@ -1041,7 +1042,7 @@ ACTOR Future<Void> recoverBlobManager(BlobManagerData* bmData) {
 				if (results[rangeIdx].value.size()) {
 					// note: if the old owner is dead, we handle this in rangeAssigner
 					UID existingOwner = decodeBlobGranuleMappingValue(results[rangeIdx].value);
-					bmData->workerAssignments.insert(KeyRangeRef(granuleStartKey, granuleEndKey), existingOwner);
+					workerAssignments.insert(KeyRangeRef(granuleStartKey, granuleEndKey), existingOwner);
 				}
 			}
 
@@ -1079,7 +1080,7 @@ ACTOR Future<Void> recoverBlobManager(BlobManagerData* bmData) {
 				std::tie(splitState, version) = decodeBlobGranuleSplitValue(split.value);
 				const KeyRange range = blobGranuleSplitKeyRangeFor(parentGranuleID);
 				if (splitState <= BlobGranuleSplitState::Initialized) {
-					bmData->workerAssignments.insert(range, UID());
+					workerAssignments.insert(range, UID());
 				}
 			}
 
@@ -1093,16 +1094,19 @@ ACTOR Future<Void> recoverBlobManager(BlobManagerData* bmData) {
 		}
 	}
 
-	// Step 3. Send assign requests for all the granules
-	for (auto& range : bmData->workerAssignments.intersectingRanges(normalKeys)) {
-		// a second manager started before the first manager assigned any ranges so the
-		// the only range is [ - \xff\xff), so we clamp it to [ - \xff)
-		Key end = range.end() == allKeys.end ? normalKeys.end : range.end();
+	// Step 3. Send assign requests for all the granules and transfer assignments
+	// from local workerAssignments to bmData
+	for (auto& range : workerAssignments.intersectingRanges(normalKeys)) {
+		if (!range.value().present()) {
+			continue;
+		}
+
+		bmData->workerAssignments.insert(range.range(), range.value().get());
 
 		RangeAssignment raAssign;
 		raAssign.isAssign = true;
-		raAssign.worker = range.value();
-		raAssign.keyRange = KeyRangeRef(range.begin(), end);
+		raAssign.worker = range.value().get();
+		raAssign.keyRange = range.range();
 		raAssign.assign = RangeAssignmentData(AssignRequestType::Reassign);
 		bmData->rangesToAssign.send(raAssign);
 	}