Merge e78981f37ae05edaae8da2a4b39b13a7a77a9050 into 166bcd9ae29cdef22525a0cbd1ae7854c281789f

fdbclient/BackupContainerFileSystem.actor.cpp

@@ -411,6 +411,42 @@ public:
 		}
 	}
 
+	// Checks if list of sorted logfiles have the logs from snapshotBeginVersion to snapshotEndversion.
+	// Which means the sorted log files(have beginVersion and endVersion) should cover
+	// all the versions between snapshotBegingVersion and snapshotEndversion.
+	// Note: logs should be sorted
+	static bool hasContinuousLogsForSnapshot(const std::vector<LogFile>& logs,
+	                                         Version snapshotBeginVersion,
+	                                         Version snapshotEndVersion) {
+		auto it = logs.begin();
+
+		// find the first mutation log file that covers snapshotBeginVersion
+		while (it != logs.end() && it->endVersion < snapshotBeginVersion) {
+			++it;
+		}
+
+		// no log find found covering snaphostBeginVersion
+		if (it == logs.end())
+			return false;
+
+		// If an overlapping log file with snapshotBeginVersion is found
+		Version prevEnd = it->endVersion;
+		++it;
+
+		while (it != logs.end()) {
+			if (it->beginVersion == prevEnd &&
+			    it->endVersion > snapshotEndVersion) // continuous logs until snapshot is covered
+				return true;
+			else if (it->beginVersion != prevEnd) // not continuous logs
+				return false;
+
+			prevEnd = it->endVersion;
+			++it;
+		} // comes out if the logs are not found until snapshotEndVersion.
+
+		return prevEnd >= snapshotEndVersion;
+	}
+
 	ACTOR static Future<BackupDescription> describeBackup(Reference<BackupContainerFileSystem> bc,
 	                                                      bool deepScan,
 	                                                      Version logStartVersionOverride) {
@@ -613,6 +649,14 @@ public:
 					s.restorable = false;
 				if (!desc.contiguousLogEnd.present() || desc.contiguousLogEnd.get() <= s.endVersion)
 					s.restorable = false;
+				// If there is a mutation logs gap after contiguousLogEnd, then check whether the current snapshot
+				// can be restored from the logs availale after contiguousLogEnd.
+				if (desc.contiguousLogEnd.present() && desc.contiguousLogEnd.get() <= s.beginVersion) {
+					if (desc.partitioned)
+						s.restorable = isPartitionedLogsContinuous(logs, s.beginVersion, s.endVersion);
+					else
+						s.restorable = hasContinuousLogsForSnapshot(logs, s.beginVersion, s.endVersion);
+				}
 			}
 
 			desc.snapshotBytes += s.totalSize;