diff --git a/fdbcli/BulkLoadCommand.cpp b/fdbcli/BulkLoadCommand.cpp
index 3bfb1a64f8a..e40196aa99e 100644
--- a/fdbcli/BulkLoadCommand.cpp
+++ b/fdbcli/BulkLoadCommand.cpp
@@ -21,6 +21,7 @@
 #include "fdbcli/fdbcli.h"
 #include "fdbclient/ManagementAPI.h"
 #include "fdbclient/BulkLoading.h"
+#include "fdbclient/RangeLock.h"
 #include "flow/Arena.h"
 #include "flow/IRandom.h"
 #include "flow/ThreadHelper.actor.h"
diff --git a/fdbcli/FileConfigureCommand.cpp b/fdbcli/FileConfigureCommand.cpp
index b4535e53686..ccad61a9496 100644
--- a/fdbcli/FileConfigureCommand.cpp
+++ b/fdbcli/FileConfigureCommand.cpp
@@ -26,6 +26,7 @@
 #include "fdbclient/IClientApi.h"
 #include "fdbclient/ManagementAPI.h"
 #include "fdbclient/Schemas.h"
+#include "fdbclient/StatusSchema.h"
 
 #include "flow/Arena.h"
 #include "flow/FastRef.h"
diff --git a/fdbcli/RangeLockCommand.cpp b/fdbcli/RangeLockCommand.cpp
index 08c162c7c30..0db65611143 100644
--- a/fdbcli/RangeLockCommand.cpp
+++ b/fdbcli/RangeLockCommand.cpp
@@ -19,7 +19,6 @@
  */
 
 #include "fdbcli/fdbcli.h"
-#include "fdbclient/ManagementAPI.h"
 #include "fdbclient/RangeLock.h"
 #include "flow/Arena.h"
 
diff --git a/fdbcli/Util.cpp b/fdbcli/Util.cpp
index 39da058925d..2c366435b4a 100644
--- a/fdbcli/Util.cpp
+++ b/fdbcli/Util.cpp
@@ -22,6 +22,7 @@
 #include "fdbclient/ManagementAPI.h"
 #include "fdbclient/Schemas.h"
 #include "fdbclient/Status.h"
+#include "fdbclient/StatusSchema.h"
 #include "fdbclient/BulkDumping.h"
 #include "fdbclient/BulkLoading.h"
 #include "flow/Arena.h"
diff --git a/fdbclient/FileBackupAgent.cpp b/fdbclient/FileBackupAgent.cpp
index 345914430b6..d98a0a3ae4d 100644
--- a/fdbclient/FileBackupAgent.cpp
+++ b/fdbclient/FileBackupAgent.cpp
@@ -39,6 +39,7 @@
 #include "fdbclient/KeyRangeMap.h"
 #include "fdbclient/Knobs.h"
 #include "fdbclient/ManagementAPI.h"
+#include "fdbclient/RangeLock.h"
 #include "PartitionedLogIterator.h"
 #include "RestoreInterface.h"
 #include "fdbclient/Status.h"
diff --git a/fdbclient/ManagementAPI.cpp b/fdbclient/ManagementAPI.cpp
index 0fc34484f99..00393653c26 100644
--- a/fdbclient/ManagementAPI.cpp
+++ b/fdbclient/ManagementAPI.cpp
@@ -48,7 +48,6 @@
 #include "flow/UnitTest.h"
 #include "fdbrpc/ReplicationPolicy.h"
 #include "fdbrpc/Replication.h"
-#include "fdbclient/Schemas.h"
 #include "fdbrpc/SimulatorProcessInfo.h"
 
 #include "flow/CoroUtils.h"
@@ -3568,389 +3567,6 @@ Future<Optional<BulkLoadProgress>> getBulkLoadProgress(Database cx) {
 	co_return progress;
 }
 
-// Persist a new owner if input ownerUniqueID is not existing; Update description if input ownerUniqueID exists
-Future<Void> registerRangeLockOwner(Database cx, RangeLockOwnerName ownerUniqueID, std::string description) {
-	if (ownerUniqueID.empty() || description.empty()) {
-		throw range_lock_failed();
-	}
-	Transaction tr(cx);
-	while (true) {
-		Error err;
-		try {
-			tr.setOption(FDBTransactionOptions::LOCK_AWARE);
-			tr.setOption(FDBTransactionOptions::ACCESS_SYSTEM_KEYS);
-			Optional<Value> res = co_await tr.get(rangeLockOwnerKeyFor(ownerUniqueID));
-			RangeLockOwner owner;
-			if (res.present()) {
-				owner = decodeRangeLockOwner(res.get());
-				ASSERT(owner.isValid());
-				if (owner.getDescription() == description) {
-					co_return;
-				}
-				owner.setDescription(description);
-			} else {
-				owner = RangeLockOwner(ownerUniqueID, description);
-			}
-			tr.set(rangeLockOwnerKeyFor(ownerUniqueID), rangeLockOwnerValue(owner));
-			co_await tr.commit();
-			co_return;
-		} catch (Error& e) {
-			err = e;
-		}
-		co_await tr.onError(err);
-	}
-}
-
-Future<Void> removeRangeLockOwner(Database cx, RangeLockOwnerName ownerUniqueID) {
-	if (ownerUniqueID.empty()) {
-		throw range_lock_failed();
-	}
-	Transaction tr(cx);
-	while (true) {
-		Error err;
-		try {
-			tr.setOption(FDBTransactionOptions::LOCK_AWARE);
-			tr.setOption(FDBTransactionOptions::ACCESS_SYSTEM_KEYS);
-			Optional<Value> res = co_await tr.get(rangeLockOwnerKeyFor(ownerUniqueID));
-			if (!res.present()) {
-				co_return;
-			}
-			RangeLockOwner owner = decodeRangeLockOwner(res.get());
-			ASSERT(owner.isValid());
-			tr.clear(rangeLockOwnerKeyFor(ownerUniqueID));
-			co_await tr.commit();
-			co_return;
-		} catch (Error& e) {
-			err = e;
-		}
-		co_await tr.onError(err);
-	}
-}
-
-Future<Optional<RangeLockOwner>> getRangeLockOwner(Database cx, RangeLockOwnerName ownerUniqueID) {
-	Transaction tr(cx);
-	while (true) {
-		Error err;
-		try {
-			tr.setOption(FDBTransactionOptions::READ_LOCK_AWARE);
-			tr.setOption(FDBTransactionOptions::READ_SYSTEM_KEYS);
-			Optional<Value> res = co_await tr.get(rangeLockOwnerKeyFor(ownerUniqueID));
-			if (!res.present()) {
-				co_return Optional<RangeLockOwner>();
-			}
-			RangeLockOwner owner = decodeRangeLockOwner(res.get());
-			ASSERT(owner.isValid());
-			co_return owner;
-		} catch (Error& e) {
-			err = e;
-		}
-		co_await tr.onError(err);
-	}
-}
-
-AsyncResult<std::vector<RangeLockOwner>> getAllRangeLockOwners(Database cx) {
-	std::vector<RangeLockOwner> res;
-	Key beginKey = rangeLockOwnerKeys.begin;
-	Key endKey = rangeLockOwnerKeys.end;
-	Transaction tr(cx);
-	while (beginKey < endKey) {
-		KeyRange rangeToRead = Standalone(KeyRangeRef(beginKey, endKey));
-		Error err;
-		try {
-			tr.setOption(FDBTransactionOptions::READ_LOCK_AWARE);
-			tr.setOption(FDBTransactionOptions::READ_SYSTEM_KEYS);
-			RangeResult result = co_await tr.getRange(rangeToRead, CLIENT_KNOBS->TOO_MANY);
-			for (const auto& kv : result) {
-				RangeLockOwner owner = decodeRangeLockOwner(kv.value);
-				ASSERT(owner.isValid());
-				res.push_back(owner);
-				beginKey = keyAfter(kv.key);
-			}
-			if (!result.more) {
-				break;
-			}
-			continue;
-		} catch (Error& e) {
-			err = e;
-		}
-		co_await tr.onError(err);
-	}
-	co_return res;
-}
-
-// Not transactional
-Future<std::vector<std::pair<KeyRange, RangeLockState>>>
-findExclusiveReadLockOnRange(Database cx, KeyRange range, Optional<RangeLockOwnerName> ownerName) {
-	if (range.end > normalKeys.end) {
-		throw range_lock_failed();
-	}
-	std::vector<std::pair<KeyRange, RangeLockState>> lockedRanges;
-	Key beginKey = range.begin;
-	Key endKey = range.end;
-	Transaction tr(cx);
-	while (beginKey < endKey) {
-		KeyRange rangeToRead = Standalone(KeyRangeRef(beginKey, endKey));
-		Error err;
-		try {
-			tr.setOption(FDBTransactionOptions::READ_LOCK_AWARE);
-			tr.setOption(FDBTransactionOptions::READ_SYSTEM_KEYS);
-			RangeResult result = co_await krmGetRanges(&tr, rangeLockPrefix, rangeToRead);
-			if (result.empty()) {
-				break;
-			}
-			for (int i = 0; i < static_cast<int>(result.size()) - 1; i++) {
-				if (result[i].value.empty()) {
-					continue;
-				}
-				RangeLockStateSet rangeLockStateSet = decodeRangeLockStateSet(result[i].value);
-				ASSERT(rangeLockStateSet.isValid());
-				if (rangeLockStateSet.isLockedFor(RangeLockType::ExclusiveReadLock) &&
-				    (!ownerName.present() ||
-				     ownerName.get() == rangeLockStateSet.getAllLockStats()[0].getOwnerUniqueId())) {
-					// Exclusive lock can only have one lock in the set, so we just check the first lock in the set
-					lockedRanges.push_back(std::make_pair(Standalone(KeyRangeRef(result[i].key, result[i + 1].key)),
-					                                      rangeLockStateSet.getAllLockStats()[0]));
-				}
-			}
-			beginKey = result.back().key;
-			continue;
-		} catch (Error& e) {
-			err = e;
-		}
-		co_await tr.onError(err);
-	}
-	co_return lockedRanges;
-}
-
-// Validate the input range and owner.
-// If invalid, reject the request by throwing range_lock_failed error.
-// If the range has been locked, reject the request by throwing range_lock_reject error.
-Future<Void> prepareExclusiveRangeLockOperation(Transaction* tr, KeyRange range, RangeLockOwnerName ownerUniqueID) {
-	// Check input range
-	if (range.end > normalKeys.end) {
-		TraceEvent(SevDebug, "PrepareExclusiveRangeLockOperationFailed")
-		    .detail("Reason", "Range out of scope")
-		    .detail("Range", range);
-		throw range_lock_failed();
-	}
-	// Check owner
-	Optional<Value> ownerValue = co_await tr->get(rangeLockOwnerKeyFor(ownerUniqueID));
-	if (!ownerValue.present()) {
-		TraceEvent(SevDebug, "PrepareExclusiveRangeLockOperationFailed")
-		    .detail("Reason", "Owner not found")
-		    .detail("Owner", ownerUniqueID)
-		    .detail("Range", range);
-		throw range_lock_failed();
-	}
-	RangeLockOwner owner = decodeRangeLockOwner(ownerValue.get());
-	ASSERT(owner.isValid());
-	// Check lock state on the entire input range. Throw exception if the range has been locked by a different owner.
-	Key beginKey = range.begin;
-	Key endKey = range.end;
-	KeyRange rangeToRead;
-	while (beginKey < endKey) {
-		rangeToRead = KeyRangeRef(beginKey, endKey);
-		RangeResult res = co_await krmGetRanges(tr, rangeLockPrefix, rangeToRead);
-		if (res.empty()) {
-			break;
-		}
-		for (int i = 0; i < static_cast<int>(res.size()) - 1; i++) {
-			if (res[i].value.empty()) {
-				continue;
-			}
-			RangeLockStateSet rangeLockStateSet = decodeRangeLockStateSet(res[i].value);
-			ASSERT(rangeLockStateSet.isValid());
-			auto lockSet = rangeLockStateSet.getLocks();
-			if (!lockSet.empty() && (!rangeLockStateSet.isLockedFor(RangeLockType::ExclusiveReadLock) ||
-			                         lockSet.find(RangeLockState(RangeLockType::ExclusiveReadLock, ownerUniqueID, range)
-			                                          .getLockUniqueString()) == lockSet.end())) {
-				TraceEvent(SevDebug, "PrepareExclusiveRangeLockOperationFailed")
-				    .detail("Reason", "Locked")
-				    .detail("NewLockType", RangeLockType::ExclusiveReadLock)
-				    .detail("NewLockRange", range)
-				    .detail("NewLockOwner", ownerUniqueID)
-				    .detail("ExistingLocks", rangeLockStateSet.toString());
-				throw range_lock_reject(); // Has been locked
-			}
-		}
-		beginKey = res.back().key;
-	}
-}
-
-Future<Void> prepareExclusiveRangeUnlockOperation(Transaction* tr, KeyRange range, RangeLockOwnerName ownerUniqueID) {
-	// Check input range
-	if (range.end > normalKeys.end) {
-		TraceEvent(SevDebug, "PrepareExclusiveRangeUnlockOperationFailed")
-		    .detail("Reason", "Range out of scope")
-		    .detail("Range", range);
-		throw range_lock_failed();
-	}
-	// Check owner
-	Optional<Value> ownerValue = co_await tr->get(rangeLockOwnerKeyFor(ownerUniqueID));
-	if (!ownerValue.present()) {
-		TraceEvent(SevDebug, "PrepareExclusiveRangeUnlockOperationFailed")
-		    .detail("Reason", "Owner not found")
-		    .detail("Owner", ownerUniqueID)
-		    .detail("Range", range);
-		throw range_lock_failed();
-	}
-	RangeLockOwner owner = decodeRangeLockOwner(ownerValue.get());
-	ASSERT(owner.isValid());
-
-	// Check lock state on the entire input range. Throw exception if the range has been locked by a different owner.
-	Key beginKey = range.begin;
-	Key endKey = range.end;
-	KeyRange rangeToRead;
-	while (beginKey < endKey) {
-		rangeToRead = KeyRangeRef(beginKey, endKey);
-		RangeResult res = co_await krmGetRanges(tr, rangeLockPrefix, rangeToRead);
-		if (res.empty()) {
-			break;
-		}
-		for (int i = 0; i < static_cast<int>(res.size()) - 1; i++) {
-			if (res[i].value.empty()) {
-				continue;
-			}
-			RangeLockStateSet rangeLockStateSet = decodeRangeLockStateSet(res[i].value);
-			ASSERT(rangeLockStateSet.isValid());
-			auto lockSet = rangeLockStateSet.getLocks();
-			if (!lockSet.empty() && (!rangeLockStateSet.isLockedFor(RangeLockType::ExclusiveReadLock) ||
-			                         lockSet.find(RangeLockState(RangeLockType::ExclusiveReadLock, ownerUniqueID, range)
-			                                          .getLockUniqueString()) == lockSet.end())) {
-				TraceEvent(SevDebug, "PrepareExclusiveRangeUnlockOperationFailed")
-				    .detail("Reason", "Has been locked by a different user or the same user with a different range")
-				    .detail("UnLockOwner", ownerUniqueID)
-				    .detail("UnLockRange", range)
-				    .detail("ExistingLocks", rangeLockStateSet.toString());
-				throw range_unlock_reject();
-			}
-		}
-		beginKey = res.back().key;
-	}
-}
-
-// Transactional. One transaction can call takeExclusiveReadLockOnRange at most for one time.
-// This is the limitation of the krmSetRangeCoalescing.
-Future<Void> takeExclusiveReadLockOnRange(Transaction* tr, KeyRange range, RangeLockOwnerName ownerUniqueID) {
-	tr->setOption(FDBTransactionOptions::LOCK_AWARE);
-	tr->setOption(FDBTransactionOptions::ACCESS_SYSTEM_KEYS);
-	// Add conflict range
-	tr->addWriteConflictRange(range);
-	co_await prepareExclusiveRangeLockOperation(tr, range, ownerUniqueID);
-	// At this point, no lock presents on the range.
-	// Lock range by writting the range.
-	RangeLockStateSet rangeLockStateSet;
-	rangeLockStateSet.insertIfNotExist(RangeLockState(RangeLockType::ExclusiveReadLock, ownerUniqueID, range));
-	co_await krmSetRange(tr, rangeLockPrefix, range, rangeLockStateSetValue(rangeLockStateSet));
-	TraceEvent(SevInfo, "TakeExclusiveReadLockTransactionOnRange").detail("Range", range);
-}
-
-// Transactional. One transaction can call releaseExclusiveReadLockOnRange at most for one time.
-// This is the limitation of the krmSetRangeCoalescing.
-Future<Void> releaseExclusiveReadLockOnRange(Transaction* tr, KeyRange range, RangeLockOwnerName ownerUniqueID) {
-	tr->setOption(FDBTransactionOptions::LOCK_AWARE);
-	tr->setOption(FDBTransactionOptions::ACCESS_SYSTEM_KEYS);
-	co_await prepareExclusiveRangeUnlockOperation(tr, range, ownerUniqueID);
-	// At this point, no lock presents on the range.
-	// Unlock by overwiting the range.
-	co_await krmSetRangeCoalescing(tr, rangeLockPrefix, range, normalKeys, rangeLockStateSetValue(RangeLockStateSet()));
-	TraceEvent(SevInfo, "ReleaseExclusiveReadLockTransactionOnRange").detail("Range", range);
-}
-
-Future<Void> releaseExclusiveReadLockByUser(Database cx, RangeLockOwnerName ownerUniqueID) {
-	Key beginKey = normalKeys.begin;
-	Key endKey = normalKeys.end;
-	Transaction tr(cx);
-	int i = 0;
-	RangeResult result;
-	KeyRange rangeToRead;
-	RangeLockStateSet currentRangeLockStateSet;
-	KeyRange currentRange;
-	Key beginKeyToClear;
-	Key endKeyToClear;
-	while (beginKey < endKey) {
-		rangeToRead = Standalone(KeyRangeRef(beginKey, endKey));
-		Error err;
-		try {
-			tr.reset();
-			tr.setOption(FDBTransactionOptions::LOCK_AWARE);
-			tr.setOption(FDBTransactionOptions::ACCESS_SYSTEM_KEYS);
-			result.clear();
-			result = co_await krmGetRanges(&tr, rangeLockPrefix, rangeToRead);
-			if (result.empty()) {
-				break;
-			}
-			i = 0;
-			beginKeyToClear = result[0].key;
-			endKeyToClear = result[0].key; // Expanding when currentRange is valid to clear
-			for (; i < static_cast<int>(result.size()) - 1; i++) {
-				currentRange = KeyRangeRef(result[i].key, result[i + 1].key);
-				if (result[i].value.empty()) {
-					endKeyToClear = currentRange.end;
-					continue;
-				}
-				currentRangeLockStateSet = decodeRangeLockStateSet(result[i].value);
-				ASSERT(currentRangeLockStateSet.isValid());
-				if (currentRangeLockStateSet.isLockedFor(RangeLockType::ExclusiveReadLock) &&
-				    currentRangeLockStateSet.getAllLockStats()[0].getOwnerUniqueId() == ownerUniqueID) {
-					// If this range is exclusively locked by the input owner, we will clear it.
-					endKeyToClear = currentRange.end;
-					continue;
-				}
-				break;
-			}
-			if (beginKeyToClear != endKeyToClear) {
-				ASSERT(endKeyToClear > beginKeyToClear);
-				co_await krmSetRangeCoalescing(&tr,
-				                               rangeLockPrefix,
-				                               KeyRangeRef(beginKeyToClear, endKeyToClear),
-				                               normalKeys,
-				                               rangeLockStateSetValue(RangeLockStateSet()));
-				co_await tr.commit();
-			}
-			beginKey = currentRange.end; // We skip the currentRange if it is not locked by the input owner.
-			continue;
-		} catch (Error& e) {
-			err = e;
-		}
-		co_await tr.onError(err);
-	}
-}
-
-// Transactional
-Future<Void> takeExclusiveReadLockOnRange(Database cx, KeyRange range, RangeLockOwnerName ownerUniqueID) {
-	Transaction tr(cx);
-	while (true) {
-		Error err;
-		try {
-			co_await takeExclusiveReadLockOnRange(&tr, range, ownerUniqueID);
-			co_await tr.commit();
-			TraceEvent(SevInfo, "TakeExclusiveReadLockOnRange").detail("Range", range);
-			break;
-		} catch (Error& e) {
-			err = e;
-		}
-		co_await tr.onError(err);
-	}
-}
-
-// Transactional
-Future<Void> releaseExclusiveReadLockOnRange(Database cx, KeyRange range, RangeLockOwnerName ownerUniqueID) {
-	Transaction tr(cx);
-	while (true) {
-		Error err;
-		try {
-			co_await releaseExclusiveReadLockOnRange(&tr, range, ownerUniqueID);
-			co_await tr.commit();
-			TraceEvent(SevInfo, "ReleaseExclusiveReadLockOnRange").detail("Range", range);
-			break;
-		} catch (Error& e) {
-			err = e;
-		}
-		co_await tr.onError(err);
-	}
-}
-
 Future<Void> waitForPrimaryDC(Database cx, StringRef dcId) {
 	ReadYourWritesTransaction tr(cx);
 
@@ -3975,173 +3591,6 @@ Future<Void> waitForPrimaryDC(Database cx, StringRef dcId) {
 	}
 }
 
-json_spirit::Value_type normJSONType(json_spirit::Value_type type) {
-	if (type == json_spirit::int_type)
-		return json_spirit::real_type;
-	return type;
-}
-
-void schemaCoverage(std::string const& spath, bool covered) {
-	static std::map<bool, std::set<std::string>> coveredSchemaPaths;
-
-	if (coveredSchemaPaths[covered].insert(spath).second) {
-		TraceEvent ev(SevInfo, "CodeCoverage");
-		ev.detail("File", "documentation/StatusSchema.json/" + spath).detail("Line", 0);
-		if (!covered)
-			ev.detail("Covered", 0);
-	}
-}
-
-bool schemaMatch(json_spirit::mValue const& schemaValue,
-                 json_spirit::mValue const& resultValue,
-                 std::string& errorStr,
-                 Severity sev,
-                 bool checkCoverage,
-                 std::string path,
-                 std::string schemaPath) {
-	// Returns true if everything in `result` is permitted by `schema`
-	bool ok = true;
-
-	try {
-		if (normJSONType(schemaValue.type()) != normJSONType(resultValue.type())) {
-			errorStr += format("ERROR: Incorrect value type for key `%s'\n", path.c_str());
-			TraceEvent(sev, "SchemaMismatch")
-			    .detail("Path", path)
-			    .detail("SchemaType", schemaValue.type())
-			    .detail("ValueType", resultValue.type());
-			return false;
-		}
-
-		if (resultValue.type() == json_spirit::obj_type) {
-			auto& result = resultValue.get_obj();
-			auto& schema = schemaValue.get_obj();
-
-			for (auto& rkv : result) {
-				auto& key = rkv.first;
-				auto& rv = rkv.second;
-				std::string kpath = path + "." + key;
-				std::string spath = schemaPath + "." + key;
-
-				if (checkCoverage) {
-					schemaCoverage(spath);
-				}
-
-				if (!schema.contains(key)) {
-					errorStr += format("ERROR: Unknown key `%s'\n", kpath.c_str());
-					TraceEvent(sev, "SchemaMismatch").detail("Path", kpath).detail("SchemaPath", spath);
-					ok = false;
-					continue;
-				}
-				auto& sv = schema.at(key);
-
-				if (sv.type() == json_spirit::obj_type && sv.get_obj().contains("$enum")) {
-					auto& enum_values = sv.get_obj().at("$enum").get_array();
-
-					bool any_match = false;
-					for (auto& enum_item : enum_values)
-						if (enum_item == rv) {
-							any_match = true;
-							if (checkCoverage) {
-								schemaCoverage(spath + ".$enum." + enum_item.get_str());
-							}
-							break;
-						}
-					if (!any_match) {
-						errorStr += format("ERROR: Unknown value `%s' for key `%s'\n",
-						                   json_spirit::write_string(rv).c_str(),
-						                   kpath.c_str());
-						TraceEvent(sev, "SchemaMismatch")
-						    .detail("Path", kpath)
-						    .detail("SchemaEnumItems", enum_values.size())
-						    .detail("Value", json_spirit::write_string(rv));
-						if (checkCoverage) {
-							schemaCoverage(spath + ".$enum." + json_spirit::write_string(rv));
-						}
-						ok = false;
-					}
-				} else if (sv.type() == json_spirit::obj_type && sv.get_obj().contains("$map")) {
-					if (rv.type() != json_spirit::obj_type) {
-						errorStr += format("ERROR: Expected an object as the value for key `%s'\n", kpath.c_str());
-						TraceEvent(sev, "SchemaMismatch")
-						    .detail("Path", kpath)
-						    .detail("SchemaType", sv.type())
-						    .detail("ValueType", rv.type());
-						ok = false;
-						continue;
-					}
-					if (sv.get_obj().at("$map").type() != json_spirit::obj_type) {
-						continue;
-					}
-					auto& schemaVal = sv.get_obj().at("$map");
-					auto& valueObj = rv.get_obj();
-
-					if (checkCoverage) {
-						schemaCoverage(spath + ".$map");
-					}
-
-					for (auto& valuePair : valueObj) {
-						auto vpath = kpath + "[" + valuePair.first + "]";
-						auto upath = spath + ".$map";
-						if (valuePair.second.type() != json_spirit::obj_type) {
-							errorStr += format("ERROR: Expected an object for `%s'\n", vpath.c_str());
-							TraceEvent(sev, "SchemaMismatch")
-							    .detail("Path", vpath)
-							    .detail("ValueType", valuePair.second.type());
-							ok = false;
-							continue;
-						}
-						if (!schemaMatch(schemaVal, valuePair.second, errorStr, sev, checkCoverage, vpath, upath)) {
-							ok = false;
-						}
-					}
-				} else {
-					if (!schemaMatch(sv, rv, errorStr, sev, checkCoverage, kpath, spath)) {
-						ok = false;
-					}
-				}
-			}
-		} else if (resultValue.type() == json_spirit::array_type) {
-			auto& valueArray = resultValue.get_array();
-			auto& schemaArray = schemaValue.get_array();
-			if (schemaArray.empty()) {
-				// An empty schema array means that the value array is required to be empty
-				if (!valueArray.empty()) {
-					errorStr += format("ERROR: Expected an empty array for key `%s'\n", path.c_str());
-					TraceEvent(sev, "SchemaMismatch")
-					    .detail("Path", path)
-					    .detail("SchemaSize", schemaArray.size())
-					    .detail("ValueSize", valueArray.size());
-					return false;
-				}
-			} else if (schemaArray.size() == 1) {
-				// A one item schema array means that all items in the value must match the first item in the schema
-				int index = 0;
-				for (auto& valueItem : valueArray) {
-					if (!schemaMatch(schemaArray[0],
-					                 valueItem,
-					                 errorStr,
-					                 sev,
-					                 checkCoverage,
-					                 path + format("[%d]", index),
-					                 schemaPath + "[0]")) {
-						ok = false;
-					}
-					index++;
-				}
-			} else {
-				ASSERT(false); // Schema doesn't make sense
-			}
-		}
-		return ok;
-	} catch (std::exception& e) {
-		TraceEvent(SevError, "SchemaMatchException")
-		    .detail("What", e.what())
-		    .detail("Path", path)
-		    .detail("SchemaPath", schemaPath);
-		throw unknown_error();
-	}
-}
-
 std::string ManagementAPI::generateErrorMessage(const CoordinatorsResult& res) {
 	// Note: the error message here should not be changed if possible
 	// If you do change the message here,
diff --git a/fdbclient/RangeLock.cpp b/fdbclient/RangeLock.cpp
new file mode 100644
index 00000000000..086d6f3dcbe
--- /dev/null
+++ b/fdbclient/RangeLock.cpp
@@ -0,0 +1,415 @@
+/*
+ * RangeLock.cpp
+ *
+ * This source file is part of the FoundationDB open source project
+ *
+ * Copyright 2013-2026 Apple Inc. and the FoundationDB project authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include "fdbclient/RangeLock.h"
+
+#include "fdbclient/FDBOptions.g.h"
+#include "fdbclient/KeyRangeMap.h"
+#include "fdbclient/Knobs.h"
+#include "fdbclient/NativeAPI.actor.h"
+#include "fdbclient/SystemData.h"
+#include "flow/Trace.h"
+
+// Persist a new owner if input ownerUniqueID is not existing; Update description if input ownerUniqueID exists
+Future<Void> registerRangeLockOwner(Database cx, RangeLockOwnerName ownerUniqueID, std::string description) {
+	if (ownerUniqueID.empty() || description.empty()) {
+		throw range_lock_failed();
+	}
+	Transaction tr(cx);
+	while (true) {
+		Error err;
+		try {
+			tr.setOption(FDBTransactionOptions::LOCK_AWARE);
+			tr.setOption(FDBTransactionOptions::ACCESS_SYSTEM_KEYS);
+			Optional<Value> res = co_await tr.get(rangeLockOwnerKeyFor(ownerUniqueID));
+			RangeLockOwner owner;
+			if (res.present()) {
+				owner = decodeRangeLockOwner(res.get());
+				ASSERT(owner.isValid());
+				if (owner.getDescription() == description) {
+					co_return;
+				}
+				owner.setDescription(description);
+			} else {
+				owner = RangeLockOwner(ownerUniqueID, description);
+			}
+			tr.set(rangeLockOwnerKeyFor(ownerUniqueID), rangeLockOwnerValue(owner));
+			co_await tr.commit();
+			co_return;
+		} catch (Error& e) {
+			err = e;
+		}
+		co_await tr.onError(err);
+	}
+}
+
+Future<Void> removeRangeLockOwner(Database cx, RangeLockOwnerName ownerUniqueID) {
+	if (ownerUniqueID.empty()) {
+		throw range_lock_failed();
+	}
+	Transaction tr(cx);
+	while (true) {
+		Error err;
+		try {
+			tr.setOption(FDBTransactionOptions::LOCK_AWARE);
+			tr.setOption(FDBTransactionOptions::ACCESS_SYSTEM_KEYS);
+			Optional<Value> res = co_await tr.get(rangeLockOwnerKeyFor(ownerUniqueID));
+			if (!res.present()) {
+				co_return;
+			}
+			RangeLockOwner owner = decodeRangeLockOwner(res.get());
+			ASSERT(owner.isValid());
+			tr.clear(rangeLockOwnerKeyFor(ownerUniqueID));
+			co_await tr.commit();
+			co_return;
+		} catch (Error& e) {
+			err = e;
+		}
+		co_await tr.onError(err);
+	}
+}
+
+Future<Optional<RangeLockOwner>> getRangeLockOwner(Database cx, RangeLockOwnerName ownerUniqueID) {
+	Transaction tr(cx);
+	while (true) {
+		Error err;
+		try {
+			tr.setOption(FDBTransactionOptions::READ_LOCK_AWARE);
+			tr.setOption(FDBTransactionOptions::READ_SYSTEM_KEYS);
+			Optional<Value> res = co_await tr.get(rangeLockOwnerKeyFor(ownerUniqueID));
+			if (!res.present()) {
+				co_return Optional<RangeLockOwner>();
+			}
+			RangeLockOwner owner = decodeRangeLockOwner(res.get());
+			ASSERT(owner.isValid());
+			co_return owner;
+		} catch (Error& e) {
+			err = e;
+		}
+		co_await tr.onError(err);
+	}
+}
+
+AsyncResult<std::vector<RangeLockOwner>> getAllRangeLockOwners(Database cx) {
+	std::vector<RangeLockOwner> res;
+	Key beginKey = rangeLockOwnerKeys.begin;
+	Key endKey = rangeLockOwnerKeys.end;
+	Transaction tr(cx);
+	while (beginKey < endKey) {
+		KeyRange rangeToRead = Standalone(KeyRangeRef(beginKey, endKey));
+		Error err;
+		try {
+			tr.setOption(FDBTransactionOptions::READ_LOCK_AWARE);
+			tr.setOption(FDBTransactionOptions::READ_SYSTEM_KEYS);
+			RangeResult result = co_await tr.getRange(rangeToRead, CLIENT_KNOBS->TOO_MANY);
+			for (const auto& kv : result) {
+				RangeLockOwner owner = decodeRangeLockOwner(kv.value);
+				ASSERT(owner.isValid());
+				res.push_back(owner);
+				beginKey = keyAfter(kv.key);
+			}
+			if (!result.more) {
+				break;
+			}
+			continue;
+		} catch (Error& e) {
+			err = e;
+		}
+		co_await tr.onError(err);
+	}
+	co_return res;
+}
+
+// Not transactional
+Future<std::vector<std::pair<KeyRange, RangeLockState>>>
+findExclusiveReadLockOnRange(Database cx, KeyRange range, Optional<RangeLockOwnerName> ownerName) {
+	if (range.end > normalKeys.end) {
+		throw range_lock_failed();
+	}
+	std::vector<std::pair<KeyRange, RangeLockState>> lockedRanges;
+	Key beginKey = range.begin;
+	Key endKey = range.end;
+	Transaction tr(cx);
+	while (beginKey < endKey) {
+		KeyRange rangeToRead = Standalone(KeyRangeRef(beginKey, endKey));
+		Error err;
+		try {
+			tr.setOption(FDBTransactionOptions::READ_LOCK_AWARE);
+			tr.setOption(FDBTransactionOptions::READ_SYSTEM_KEYS);
+			RangeResult result = co_await krmGetRanges(&tr, rangeLockPrefix, rangeToRead);
+			if (result.empty()) {
+				break;
+			}
+			for (int i = 0; i < static_cast<int>(result.size()) - 1; i++) {
+				if (result[i].value.empty()) {
+					continue;
+				}
+				RangeLockStateSet rangeLockStateSet = decodeRangeLockStateSet(result[i].value);
+				ASSERT(rangeLockStateSet.isValid());
+				if (rangeLockStateSet.isLockedFor(RangeLockType::ExclusiveReadLock) &&
+				    (!ownerName.present() ||
+				     ownerName.get() == rangeLockStateSet.getAllLockStats()[0].getOwnerUniqueId())) {
+					// Exclusive lock can only have one lock in the set, so we just check the first lock in the set
+					lockedRanges.push_back(std::make_pair(Standalone(KeyRangeRef(result[i].key, result[i + 1].key)),
+					                                      rangeLockStateSet.getAllLockStats()[0]));
+				}
+			}
+			beginKey = result.back().key;
+			continue;
+		} catch (Error& e) {
+			err = e;
+		}
+		co_await tr.onError(err);
+	}
+	co_return lockedRanges;
+}
+
+namespace {
+
+// Validate the input range and owner.
+// If invalid, reject the request by throwing range_lock_failed error.
+// If the range has been locked, reject the request by throwing range_lock_reject error.
+Future<Void> prepareExclusiveRangeLockOperation(Transaction* tr, KeyRange range, RangeLockOwnerName ownerUniqueID) {
+	// Check input range
+	if (range.end > normalKeys.end) {
+		TraceEvent(SevDebug, "PrepareExclusiveRangeLockOperationFailed")
+		    .detail("Reason", "Range out of scope")
+		    .detail("Range", range);
+		throw range_lock_failed();
+	}
+	// Check owner
+	Optional<Value> ownerValue = co_await tr->get(rangeLockOwnerKeyFor(ownerUniqueID));
+	if (!ownerValue.present()) {
+		TraceEvent(SevDebug, "PrepareExclusiveRangeLockOperationFailed")
+		    .detail("Reason", "Owner not found")
+		    .detail("Owner", ownerUniqueID)
+		    .detail("Range", range);
+		throw range_lock_failed();
+	}
+	RangeLockOwner owner = decodeRangeLockOwner(ownerValue.get());
+	ASSERT(owner.isValid());
+	// Check lock state on the entire input range. Throw exception if the range has been locked by a different owner.
+	Key beginKey = range.begin;
+	Key endKey = range.end;
+	KeyRange rangeToRead;
+	while (beginKey < endKey) {
+		rangeToRead = KeyRangeRef(beginKey, endKey);
+		RangeResult res = co_await krmGetRanges(tr, rangeLockPrefix, rangeToRead);
+		if (res.empty()) {
+			break;
+		}
+		for (int i = 0; i < static_cast<int>(res.size()) - 1; i++) {
+			if (res[i].value.empty()) {
+				continue;
+			}
+			RangeLockStateSet rangeLockStateSet = decodeRangeLockStateSet(res[i].value);
+			ASSERT(rangeLockStateSet.isValid());
+			auto lockSet = rangeLockStateSet.getLocks();
+			if (!lockSet.empty() && (!rangeLockStateSet.isLockedFor(RangeLockType::ExclusiveReadLock) ||
+			                         lockSet.find(RangeLockState(RangeLockType::ExclusiveReadLock, ownerUniqueID, range)
+			                                          .getLockUniqueString()) == lockSet.end())) {
+				TraceEvent(SevDebug, "PrepareExclusiveRangeLockOperationFailed")
+				    .detail("Reason", "Locked")
+				    .detail("NewLockType", RangeLockType::ExclusiveReadLock)
+				    .detail("NewLockRange", range)
+				    .detail("NewLockOwner", ownerUniqueID)
+				    .detail("ExistingLocks", rangeLockStateSet.toString());
+				throw range_lock_reject(); // Has been locked
+			}
+		}
+		beginKey = res.back().key;
+	}
+}
+
+Future<Void> prepareExclusiveRangeUnlockOperation(Transaction* tr, KeyRange range, RangeLockOwnerName ownerUniqueID) {
+	// Check input range
+	if (range.end > normalKeys.end) {
+		TraceEvent(SevDebug, "PrepareExclusiveRangeUnlockOperationFailed")
+		    .detail("Reason", "Range out of scope")
+		    .detail("Range", range);
+		throw range_lock_failed();
+	}
+	// Check owner
+	Optional<Value> ownerValue = co_await tr->get(rangeLockOwnerKeyFor(ownerUniqueID));
+	if (!ownerValue.present()) {
+		TraceEvent(SevDebug, "PrepareExclusiveRangeUnlockOperationFailed")
+		    .detail("Reason", "Owner not found")
+		    .detail("Owner", ownerUniqueID)
+		    .detail("Range", range);
+		throw range_lock_failed();
+	}
+	RangeLockOwner owner = decodeRangeLockOwner(ownerValue.get());
+	ASSERT(owner.isValid());
+
+	// Check lock state on the entire input range. Throw exception if the range has been locked by a different owner.
+	Key beginKey = range.begin;
+	Key endKey = range.end;
+	KeyRange rangeToRead;
+	while (beginKey < endKey) {
+		rangeToRead = KeyRangeRef(beginKey, endKey);
+		RangeResult res = co_await krmGetRanges(tr, rangeLockPrefix, rangeToRead);
+		if (res.empty()) {
+			break;
+		}
+		for (int i = 0; i < static_cast<int>(res.size()) - 1; i++) {
+			if (res[i].value.empty()) {
+				continue;
+			}
+			RangeLockStateSet rangeLockStateSet = decodeRangeLockStateSet(res[i].value);
+			ASSERT(rangeLockStateSet.isValid());
+			auto lockSet = rangeLockStateSet.getLocks();
+			if (!lockSet.empty() && (!rangeLockStateSet.isLockedFor(RangeLockType::ExclusiveReadLock) ||
+			                         lockSet.find(RangeLockState(RangeLockType::ExclusiveReadLock, ownerUniqueID, range)
+			                                          .getLockUniqueString()) == lockSet.end())) {
+				TraceEvent(SevDebug, "PrepareExclusiveRangeUnlockOperationFailed")
+				    .detail("Reason", "Has been locked by a different user or the same user with a different range")
+				    .detail("UnLockOwner", ownerUniqueID)
+				    .detail("UnLockRange", range)
+				    .detail("ExistingLocks", rangeLockStateSet.toString());
+				throw range_unlock_reject();
+			}
+		}
+		beginKey = res.back().key;
+	}
+}
+
+} // namespace
+
+// Transactional. One transaction can call takeExclusiveReadLockOnRange at most for one time.
+// This is the limitation of the krmSetRangeCoalescing.
+Future<Void> takeExclusiveReadLockOnRange(Transaction* tr, KeyRange range, RangeLockOwnerName ownerUniqueID) {
+	tr->setOption(FDBTransactionOptions::LOCK_AWARE);
+	tr->setOption(FDBTransactionOptions::ACCESS_SYSTEM_KEYS);
+	// Add conflict range
+	tr->addWriteConflictRange(range);
+	co_await prepareExclusiveRangeLockOperation(tr, range, ownerUniqueID);
+	// At this point, no lock presents on the range.
+	// Lock range by writting the range.
+	RangeLockStateSet rangeLockStateSet;
+	rangeLockStateSet.insertIfNotExist(RangeLockState(RangeLockType::ExclusiveReadLock, ownerUniqueID, range));
+	co_await krmSetRange(tr, rangeLockPrefix, range, rangeLockStateSetValue(rangeLockStateSet));
+	TraceEvent(SevInfo, "TakeExclusiveReadLockTransactionOnRange").detail("Range", range);
+}
+
+// Transactional. One transaction can call releaseExclusiveReadLockOnRange at most for one time.
+// This is the limitation of the krmSetRangeCoalescing.
+Future<Void> releaseExclusiveReadLockOnRange(Transaction* tr, KeyRange range, RangeLockOwnerName ownerUniqueID) {
+	tr->setOption(FDBTransactionOptions::LOCK_AWARE);
+	tr->setOption(FDBTransactionOptions::ACCESS_SYSTEM_KEYS);
+	co_await prepareExclusiveRangeUnlockOperation(tr, range, ownerUniqueID);
+	// At this point, no lock presents on the range.
+	// Unlock by overwiting the range.
+	co_await krmSetRangeCoalescing(tr, rangeLockPrefix, range, normalKeys, rangeLockStateSetValue(RangeLockStateSet()));
+	TraceEvent(SevInfo, "ReleaseExclusiveReadLockTransactionOnRange").detail("Range", range);
+}
+
+Future<Void> releaseExclusiveReadLockByUser(Database cx, RangeLockOwnerName ownerUniqueID) {
+	Key beginKey = normalKeys.begin;
+	Key endKey = normalKeys.end;
+	Transaction tr(cx);
+	int i = 0;
+	RangeResult result;
+	KeyRange rangeToRead;
+	RangeLockStateSet currentRangeLockStateSet;
+	KeyRange currentRange;
+	Key beginKeyToClear;
+	Key endKeyToClear;
+	while (beginKey < endKey) {
+		rangeToRead = Standalone(KeyRangeRef(beginKey, endKey));
+		Error err;
+		try {
+			tr.reset();
+			tr.setOption(FDBTransactionOptions::LOCK_AWARE);
+			tr.setOption(FDBTransactionOptions::ACCESS_SYSTEM_KEYS);
+			result.clear();
+			result = co_await krmGetRanges(&tr, rangeLockPrefix, rangeToRead);
+			if (result.empty()) {
+				break;
+			}
+			i = 0;
+			beginKeyToClear = result[0].key;
+			endKeyToClear = result[0].key; // Expanding when currentRange is valid to clear
+			for (; i < static_cast<int>(result.size()) - 1; i++) {
+				currentRange = KeyRangeRef(result[i].key, result[i + 1].key);
+				if (result[i].value.empty()) {
+					endKeyToClear = currentRange.end;
+					continue;
+				}
+				currentRangeLockStateSet = decodeRangeLockStateSet(result[i].value);
+				ASSERT(currentRangeLockStateSet.isValid());
+				if (currentRangeLockStateSet.isLockedFor(RangeLockType::ExclusiveReadLock) &&
+				    currentRangeLockStateSet.getAllLockStats()[0].getOwnerUniqueId() == ownerUniqueID) {
+					// If this range is exclusively locked by the input owner, we will clear it.
+					endKeyToClear = currentRange.end;
+					continue;
+				}
+				break;
+			}
+			if (beginKeyToClear != endKeyToClear) {
+				ASSERT(endKeyToClear > beginKeyToClear);
+				co_await krmSetRangeCoalescing(&tr,
+				                               rangeLockPrefix,
+				                               KeyRangeRef(beginKeyToClear, endKeyToClear),
+				                               normalKeys,
+				                               rangeLockStateSetValue(RangeLockStateSet()));
+				co_await tr.commit();
+			}
+			beginKey = currentRange.end; // We skip the currentRange if it is not locked by the input owner.
+			continue;
+		} catch (Error& e) {
+			err = e;
+		}
+		co_await tr.onError(err);
+	}
+}
+
+// Transactional
+Future<Void> takeExclusiveReadLockOnRange(Database cx, KeyRange range, RangeLockOwnerName ownerUniqueID) {
+	Transaction tr(cx);
+	while (true) {
+		Error err;
+		try {
+			co_await takeExclusiveReadLockOnRange(&tr, range, ownerUniqueID);
+			co_await tr.commit();
+			TraceEvent(SevInfo, "TakeExclusiveReadLockOnRange").detail("Range", range);
+			break;
+		} catch (Error& e) {
+			err = e;
+		}
+		co_await tr.onError(err);
+	}
+}
+
+// Transactional
+Future<Void> releaseExclusiveReadLockOnRange(Database cx, KeyRange range, RangeLockOwnerName ownerUniqueID) {
+	Transaction tr(cx);
+	while (true) {
+		Error err;
+		try {
+			co_await releaseExclusiveReadLockOnRange(&tr, range, ownerUniqueID);
+			co_await tr.commit();
+			TraceEvent(SevInfo, "ReleaseExclusiveReadLockOnRange").detail("Range", range);
+			break;
+		} catch (Error& e) {
+			err = e;
+		}
+		co_await tr.onError(err);
+	}
+}
diff --git a/fdbclient/StatusSchema.cpp b/fdbclient/StatusSchema.cpp
new file mode 100644
index 00000000000..4d419801a86
--- /dev/null
+++ b/fdbclient/StatusSchema.cpp
@@ -0,0 +1,200 @@
+/*
+ * StatusSchema.cpp
+ *
+ * This source file is part of the FoundationDB open source project
+ *
+ * Copyright 2013-2026 Apple Inc. and the FoundationDB project authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include "fdbclient/StatusSchema.h"
+
+#include <exception>
+#include <map>
+#include <set>
+
+#include "fdbclient/json_spirit/json_spirit_writer_template.h"
+#include "flow/Error.h"
+#include "fmt/format.h"
+
+namespace {
+
+json_spirit::Value_type normJSONType(json_spirit::Value_type type) {
+	if (type == json_spirit::int_type)
+		return json_spirit::real_type;
+	return type;
+}
+
+} // namespace
+
+void schemaCoverage(std::string const& spath, bool covered) {
+	static std::map<bool, std::set<std::string>> coveredSchemaPaths;
+
+	if (coveredSchemaPaths[covered].insert(spath).second) {
+		TraceEvent ev(SevInfo, "CodeCoverage");
+		ev.detail("File", "documentation/StatusSchema.json/" + spath).detail("Line", 0);
+		if (!covered)
+			ev.detail("Covered", 0);
+	}
+}
+
+bool schemaMatch(json_spirit::mValue const& schemaValue,
+                 json_spirit::mValue const& resultValue,
+                 std::string& errorStr,
+                 Severity sev,
+                 bool checkCoverage,
+                 std::string path,
+                 std::string schemaPath) {
+	// Returns true if everything in `result` is permitted by `schema`
+	bool ok = true;
+
+	try {
+		if (normJSONType(schemaValue.type()) != normJSONType(resultValue.type())) {
+			errorStr += format("ERROR: Incorrect value type for key `%s'\n", path.c_str());
+			TraceEvent(sev, "SchemaMismatch")
+			    .detail("Path", path)
+			    .detail("SchemaType", schemaValue.type())
+			    .detail("ValueType", resultValue.type());
+			return false;
+		}
+
+		if (resultValue.type() == json_spirit::obj_type) {
+			auto& result = resultValue.get_obj();
+			auto& schema = schemaValue.get_obj();
+
+			for (auto& rkv : result) {
+				auto& key = rkv.first;
+				auto& rv = rkv.second;
+				std::string kpath = path + "." + key;
+				std::string spath = schemaPath + "." + key;
+
+				if (checkCoverage) {
+					schemaCoverage(spath);
+				}
+
+				if (!schema.contains(key)) {
+					errorStr += format("ERROR: Unknown key `%s'\n", kpath.c_str());
+					TraceEvent(sev, "SchemaMismatch").detail("Path", kpath).detail("SchemaPath", spath);
+					ok = false;
+					continue;
+				}
+				auto& sv = schema.at(key);
+
+				if (sv.type() == json_spirit::obj_type && sv.get_obj().contains("$enum")) {
+					auto& enum_values = sv.get_obj().at("$enum").get_array();
+
+					bool any_match = false;
+					for (auto& enum_item : enum_values)
+						if (enum_item == rv) {
+							any_match = true;
+							if (checkCoverage) {
+								schemaCoverage(spath + ".$enum." + enum_item.get_str());
+							}
+							break;
+						}
+					if (!any_match) {
+						errorStr += format("ERROR: Unknown value `%s' for key `%s'\n",
+						                   json_spirit::write_string(rv).c_str(),
+						                   kpath.c_str());
+						TraceEvent(sev, "SchemaMismatch")
+						    .detail("Path", kpath)
+						    .detail("SchemaEnumItems", enum_values.size())
+						    .detail("Value", json_spirit::write_string(rv));
+						if (checkCoverage) {
+							schemaCoverage(spath + ".$enum." + json_spirit::write_string(rv));
+						}
+						ok = false;
+					}
+				} else if (sv.type() == json_spirit::obj_type && sv.get_obj().contains("$map")) {
+					if (rv.type() != json_spirit::obj_type) {
+						errorStr += format("ERROR: Expected an object as the value for key `%s'\n", kpath.c_str());
+						TraceEvent(sev, "SchemaMismatch")
+						    .detail("Path", kpath)
+						    .detail("SchemaType", sv.type())
+						    .detail("ValueType", rv.type());
+						ok = false;
+						continue;
+					}
+					if (sv.get_obj().at("$map").type() != json_spirit::obj_type) {
+						continue;
+					}
+					auto& schemaVal = sv.get_obj().at("$map");
+					auto& valueObj = rv.get_obj();
+
+					if (checkCoverage) {
+						schemaCoverage(spath + ".$map");
+					}
+
+					for (auto& valuePair : valueObj) {
+						auto vpath = kpath + "[" + valuePair.first + "]";
+						auto upath = spath + ".$map";
+						if (valuePair.second.type() != json_spirit::obj_type) {
+							errorStr += format("ERROR: Expected an object for `%s'\n", vpath.c_str());
+							TraceEvent(sev, "SchemaMismatch")
+							    .detail("Path", vpath)
+							    .detail("ValueType", valuePair.second.type());
+							ok = false;
+							continue;
+						}
+						if (!schemaMatch(schemaVal, valuePair.second, errorStr, sev, checkCoverage, vpath, upath)) {
+							ok = false;
+						}
+					}
+				} else {
+					if (!schemaMatch(sv, rv, errorStr, sev, checkCoverage, kpath, spath)) {
+						ok = false;
+					}
+				}
+			}
+		} else if (resultValue.type() == json_spirit::array_type) {
+			auto& valueArray = resultValue.get_array();
+			auto& schemaArray = schemaValue.get_array();
+			if (schemaArray.empty()) {
+				// An empty schema array means that the value array is required to be empty
+				if (!valueArray.empty()) {
+					errorStr += format("ERROR: Expected an empty array for key `%s'\n", path.c_str());
+					TraceEvent(sev, "SchemaMismatch")
+					    .detail("Path", path)
+					    .detail("SchemaSize", schemaArray.size())
+					    .detail("ValueSize", valueArray.size());
+					return false;
+				}
+			} else if (schemaArray.size() == 1) {
+				// A one item schema array means that all items in the value must match the first item in the schema
+				int index = 0;
+				for (auto& valueItem : valueArray) {
+					if (!schemaMatch(schemaArray[0],
+					                 valueItem,
+					                 errorStr,
+					                 sev,
+					                 checkCoverage,
+					                 path + format("[%d]", index),
+					                 schemaPath + "[0]")) {
+						ok = false;
+					}
+					index++;
+				}
+			} else {
+				ASSERT(false); // Schema doesn't make sense
+			}
+		}
+		return ok;
+	} catch (std::exception& e) {
+		TraceEvent(SevError, "SchemaMatchException")
+		    .detail("What", e.what())
+		    .detail("Path", path)
+		    .detail("SchemaPath", schemaPath);
+		throw unknown_error();
+	}
+}
diff --git a/fdbclient/include/fdbclient/ManagementAPI.h b/fdbclient/include/fdbclient/ManagementAPI.h
index 35496efda33..a24cf20ce4e 100644
--- a/fdbclient/include/fdbclient/ManagementAPI.h
+++ b/fdbclient/include/fdbclient/ManagementAPI.h
@@ -31,7 +31,6 @@ standard API and some knowledge of the contents of the system key space.
 #include <map>
 #include "fdbclient/GenericManagementAPI.h"
 #include "fdbclient/NativeAPI.actor.h"
-#include "fdbclient/RangeLock.h"
 #include "fdbclient/ReadYourWrites.h"
 #include "fdbclient/DatabaseConfiguration.h"
 #include "fdbclient/MonitorLeader.h"
@@ -357,40 +356,6 @@ Future<Optional<BulkDumpOwnerInfo>> getBulkLoadOwner(Database cx, UID jobId);
 
 // ==================== End Progress Tracking ====================
 
-// Persist a rangeLock owner to database metadata
-// A range can only be locked by a registered owner
-Future<Void> registerRangeLockOwner(Database cx, RangeLockOwnerName ownerUniqueID, std::string description);
-
-// Remove an owner form the database metadata
-Future<Void> removeRangeLockOwner(Database cx, RangeLockOwnerName ownerUniqueID);
-
-// Get all registered rangeLock owner
-AsyncResult<std::vector<RangeLockOwner>> getAllRangeLockOwners(Database cx);
-
-// Get a rangeLock owner by ownerUniqueID
-Future<Optional<RangeLockOwner>> getRangeLockOwner(Database cx, RangeLockOwnerName ownerUniqueID);
-
-// Block write traffic to a user range (the input range must be within normalKeys).
-// One transaction can call releaseExclusiveReadLockOnRange at most for one time.
-Future<Void> takeExclusiveReadLockOnRange(Transaction* tr, KeyRange range, RangeLockOwnerName ownerUniqueID);
-
-Future<Void> takeExclusiveReadLockOnRange(Database cx, KeyRange range, RangeLockOwnerName ownerUniqueID);
-
-// Unblock a user range (the input range must be within normalKeys).
-// One transaction can call releaseExclusiveReadLockOnRange at most for one time.
-Future<Void> releaseExclusiveReadLockOnRange(Transaction* tr, KeyRange range, RangeLockOwnerName ownerUniqueID);
-
-Future<Void> releaseExclusiveReadLockOnRange(Database cx, KeyRange range, RangeLockOwnerName ownerUniqueID);
-
-// Get locked ranges within the input range (the input range must be within normalKeys)
-Future<std::vector<std::pair<KeyRange, RangeLockState>>> findExclusiveReadLockOnRange(
-    Database cx,
-    KeyRange range,
-    Optional<RangeLockOwnerName> ownerName = Optional<RangeLockOwnerName>());
-
-// Clear all exclusive read lock by the input user. Not transactional.
-Future<Void> releaseExclusiveReadLockByUser(Database cx, RangeLockOwnerName ownerUniqueID);
-
 Future<Void> printHealthyZone(Database cx);
 Future<bool> clearHealthyZone(Database cx, bool printWarning = false, bool clearSSFailureZoneString = false);
 Future<bool> setHealthyZone(Database cx, StringRef zoneId, double seconds, bool printWarning = false);
@@ -400,15 +365,6 @@ Future<Void> waitForPrimaryDC(Database cx, StringRef dcId);
 // Gets the cluster connection string
 Future<Optional<ClusterConnectionString>> getConnectionString(Database cx);
 
-void schemaCoverage(std::string const& spath, bool covered = true);
-bool schemaMatch(json_spirit::mValue const& schema,
-                 json_spirit::mValue const& result,
-                 std::string& errorStr,
-                 Severity sev = SevError,
-                 bool checkCoverage = false,
-                 std::string path = std::string(),
-                 std::string schema_path = std::string());
-
 // execute payload in 'snapCmd' on all the coordinators, TLogs and
 // storage nodes
 Future<Void> mgmtSnapCreate(Database cx, Standalone<StringRef> snapCmd, UID snapUID);
diff --git a/fdbclient/include/fdbclient/RangeLock.h b/fdbclient/include/fdbclient/RangeLock.h
index 9f319e67000..744d2438759 100644
--- a/fdbclient/include/fdbclient/RangeLock.h
+++ b/fdbclient/include/fdbclient/RangeLock.h
@@ -32,6 +32,8 @@ using RangeLockOwnerName = std::string;
 using RangeLockUniqueString = std::string;
 using RangeLockID = std::string;
 
+class Transaction;
+
 enum class RangeLockType : uint8_t {
 	Invalid = 0,
 	ExclusiveReadLock = 1, // reject all commits to the locked range
@@ -222,4 +224,36 @@ struct RangeLockStateSet {
 	std::map<RangeLockUniqueString, RangeLockState> locks;
 };
 
+// Persist a rangeLock owner to database metadata.
+// A range can only be locked by a registered owner.
+Future<Void> registerRangeLockOwner(Database cx, RangeLockOwnerName ownerUniqueID, std::string description);
+
+// Remove an owner from the database metadata.
+Future<Void> removeRangeLockOwner(Database cx, RangeLockOwnerName ownerUniqueID);
+
+// Get all registered rangeLock owners.
+AsyncResult<std::vector<RangeLockOwner>> getAllRangeLockOwners(Database cx);
+
+// Get a rangeLock owner by ownerUniqueID.
+Future<Optional<RangeLockOwner>> getRangeLockOwner(Database cx, RangeLockOwnerName ownerUniqueID);
+
+// Block write traffic to a user range (the input range must be within normalKeys).
+// One transaction can call takeExclusiveReadLockOnRange at most one time.
+Future<Void> takeExclusiveReadLockOnRange(Transaction* tr, KeyRange range, RangeLockOwnerName ownerUniqueID);
+Future<Void> takeExclusiveReadLockOnRange(Database cx, KeyRange range, RangeLockOwnerName ownerUniqueID);
+
+// Unblock a user range (the input range must be within normalKeys).
+// One transaction can call releaseExclusiveReadLockOnRange at most one time.
+Future<Void> releaseExclusiveReadLockOnRange(Transaction* tr, KeyRange range, RangeLockOwnerName ownerUniqueID);
+Future<Void> releaseExclusiveReadLockOnRange(Database cx, KeyRange range, RangeLockOwnerName ownerUniqueID);
+
+// Get locked ranges within the input range (the input range must be within normalKeys).
+Future<std::vector<std::pair<KeyRange, RangeLockState>>> findExclusiveReadLockOnRange(
+    Database cx,
+    KeyRange range,
+    Optional<RangeLockOwnerName> ownerName = Optional<RangeLockOwnerName>());
+
+// Clear all exclusive read locks owned by the input user. Not transactional.
+Future<Void> releaseExclusiveReadLockByUser(Database cx, RangeLockOwnerName ownerUniqueID);
+
 #endif
diff --git a/fdbclient/include/fdbclient/StatusSchema.h b/fdbclient/include/fdbclient/StatusSchema.h
new file mode 100644
index 00000000000..dccf3e6ae98
--- /dev/null
+++ b/fdbclient/include/fdbclient/StatusSchema.h
@@ -0,0 +1,39 @@
+/*
+ * StatusSchema.h
+ *
+ * This source file is part of the FoundationDB open source project
+ *
+ * Copyright 2013-2026 Apple Inc. and the FoundationDB project authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#ifndef FDBCLIENT_STATUSSCHEMA_H
+#define FDBCLIENT_STATUSSCHEMA_H
+#pragma once
+
+#include <string>
+
+#include "fdbclient/json_spirit/json_spirit_value.h"
+#include "flow/Trace.h"
+
+void schemaCoverage(std::string const& spath, bool covered = true);
+bool schemaMatch(json_spirit::mValue const& schema,
+                 json_spirit::mValue const& result,
+                 std::string& errorStr,
+                 Severity sev = SevError,
+                 bool checkCoverage = false,
+                 std::string path = std::string(),
+                 std::string schema_path = std::string());
+
+#endif
diff --git a/fdbctl/ControlCommands.cpp b/fdbctl/ControlCommands.cpp
index 05a630c3522..6c0113baeb8 100644
--- a/fdbctl/ControlCommands.cpp
+++ b/fdbctl/ControlCommands.cpp
@@ -21,6 +21,7 @@
 #include "fdbctl/ControlCommands.h"
 #include "fdbclient/ManagementAPI.h"
 #include "fdbclient/Schemas.h"
+#include "fdbclient/StatusSchema.h"
 #include "fmt/format.h"
 #include <boost/algorithm/string.hpp>
 
diff --git a/fdbserver/core/LatencyBandConfig.cpp b/fdbserver/core/LatencyBandConfig.cpp
index 8cc4ab0e024..83b920cf06d 100644
--- a/fdbserver/core/LatencyBandConfig.cpp
+++ b/fdbserver/core/LatencyBandConfig.cpp
@@ -20,8 +20,8 @@
 
 #include "fdbserver/core/LatencyBandConfig.h"
 
-#include "fdbclient/ManagementAPI.h"
 #include "fdbclient/Schemas.h"
+#include "fdbclient/StatusSchema.h"
 
 bool operator==(LatencyBandConfig::RequestConfig const& lhs, LatencyBandConfig::RequestConfig const& rhs) {
 	return typeid(lhs) == typeid(rhs) && lhs.isEqual(rhs);
diff --git a/fdbserver/datadistributor/DataDistribution.cpp b/fdbserver/datadistributor/DataDistribution.cpp
index aeff01c04e8..07d0ee41195 100644
--- a/fdbserver/datadistributor/DataDistribution.cpp
+++ b/fdbserver/datadistributor/DataDistribution.cpp
@@ -29,6 +29,7 @@
 #include "fdbclient/FDBTypes.h"
 #include "fdbclient/Knobs.h"
 #include "fdbclient/ManagementAPI.h"
+#include "fdbclient/RangeLock.h"
 #include "fdbclient/RunRYWTransaction.h"
 #include "fdbclient/StorageServerInterface.h"
 #include "fdbclient/SystemData.h"
diff --git a/fdbserver/workloads/BulkDumping.cpp b/fdbserver/workloads/BulkDumping.cpp
index 888e5031db7..bd9daebd82e 100644
--- a/fdbserver/workloads/BulkDumping.cpp
+++ b/fdbserver/workloads/BulkDumping.cpp
@@ -23,6 +23,7 @@
 #include "fdbclient/FDBTypes.h"
 #include "fdbclient/ManagementAPI.h"
 #include "fdbclient/NativeAPI.actor.h"
+#include "fdbclient/RangeLock.h"
 #include "fdbserver/core/Knobs.h"
 #include "fdbserver/tester/workloads.h"
 #include "fdbserver/mocks3/MockS3Server.h"
diff --git a/fdbserver/workloads/BulkLoading.cpp b/fdbserver/workloads/BulkLoading.cpp
index be1360a8f9e..4ad0f167e7e 100644
--- a/fdbserver/workloads/BulkLoading.cpp
+++ b/fdbserver/workloads/BulkLoading.cpp
@@ -21,6 +21,7 @@
 #include "fdbclient/BulkLoading.h"
 #include "fdbclient/ManagementAPI.h"
 #include "fdbclient/NativeAPI.actor.h"
+#include "fdbclient/RangeLock.h"
 #include "fdbclient/SystemData.h"
 #include "fdbserver/core/BulkLoadUtil.h"
 #include "fdbserver/core/RocksDBCheckpointUtils.h"
diff --git a/fdbserver/workloads/ChangeConfig.cpp b/fdbserver/workloads/ChangeConfig.cpp
index f9ff279d3d0..9da0fa490e1 100644
--- a/fdbserver/workloads/ChangeConfig.cpp
+++ b/fdbserver/workloads/ChangeConfig.cpp
@@ -21,6 +21,7 @@
 #include "fdbclient/ManagementAPI.h"
 #include "fdbclient/NativeAPI.actor.h"
 #include "fdbclient/Schemas.h"
+#include "fdbclient/StatusSchema.h"
 #include "fdbserver/tester/workloads.h"
 #include "fdbrpc/simulator.h"
 
diff --git a/fdbserver/workloads/DataDistributionMetrics.cpp b/fdbserver/workloads/DataDistributionMetrics.cpp
index 687d2463093..4a45c87f96c 100644
--- a/fdbserver/workloads/DataDistributionMetrics.cpp
+++ b/fdbserver/workloads/DataDistributionMetrics.cpp
@@ -20,9 +20,9 @@
 
 #include <boost/lexical_cast.hpp>
 
-#include "fdbclient/ManagementAPI.h"
 #include "fdbclient/ReadYourWrites.h"
 #include "fdbclient/Schemas.h"
+#include "fdbclient/StatusSchema.h"
 #include "fdbserver/tester/workloads.h"
 
 struct DataDistributionMetricsWorkload : KVWorkload {
diff --git a/fdbserver/workloads/RandomRangeLock.cpp b/fdbserver/workloads/RandomRangeLock.cpp
index 3f34a1efe9b..87d16593e29 100644
--- a/fdbserver/workloads/RandomRangeLock.cpp
+++ b/fdbserver/workloads/RandomRangeLock.cpp
@@ -19,7 +19,6 @@
  */
 
 #include "fdbclient/FDBTypes.h"
-#include "fdbclient/ManagementAPI.h"
 #include "fdbclient/RangeLock.h"
 #include "fdbclient/SystemData.h"
 #include "fdbserver/core/Knobs.h"
diff --git a/fdbserver/workloads/RangeLock.cpp b/fdbserver/workloads/RangeLock.cpp
index 37b8fe43b5d..5625811c87f 100644
--- a/fdbserver/workloads/RangeLock.cpp
+++ b/fdbserver/workloads/RangeLock.cpp
@@ -21,7 +21,6 @@
 #include "fdbclient/AuditUtils.h"
 #include "fdbclient/RangeLock.h"
 #include "fdbclient/FDBTypes.h"
-#include "fdbclient/ManagementAPI.h"
 #include "fdbclient/SystemData.h"
 #include "fdbserver/core/TesterInterface.h"
 #include "fdbserver/tester/workloads.h"
diff --git a/fdbserver/workloads/SpecialKeySpaceCorrectness.cpp b/fdbserver/workloads/SpecialKeySpaceCorrectness.cpp
index 1e6563fab9d..364d2aacd55 100644
--- a/fdbserver/workloads/SpecialKeySpaceCorrectness.cpp
+++ b/fdbserver/workloads/SpecialKeySpaceCorrectness.cpp
@@ -27,6 +27,7 @@
 #include "fdbclient/ReadYourWrites.h"
 #include "fdbclient/Schemas.h"
 #include "fdbclient/SpecialKeySpace.h"
+#include "fdbclient/StatusSchema.h"
 #include "fdbserver/core/Knobs.h"
 #include "fdbserver/core/TesterInterface.h"
 #include "fdbserver/tester/workloads.h"
diff --git a/fdbserver/workloads/SpecialKeySpaceRobustness.cpp b/fdbserver/workloads/SpecialKeySpaceRobustness.cpp
index 4a95555d8fd..6beaadc820e 100644
--- a/fdbserver/workloads/SpecialKeySpaceRobustness.cpp
+++ b/fdbserver/workloads/SpecialKeySpaceRobustness.cpp
@@ -25,6 +25,7 @@
 #include "fdbclient/ReadYourWrites.h"
 #include "fdbclient/Schemas.h"
 #include "fdbclient/SpecialKeySpace.h"
+#include "fdbclient/StatusSchema.h"
 #include "fdbserver/tester/workloads.h"
 
 struct SpecialKeySpaceRobustnessWorkload : TestWorkload {
diff --git a/fdbserver/workloads/StatusWorkload.cpp b/fdbserver/workloads/StatusWorkload.cpp
index 09d195fcae7..1161f0811e9 100644
--- a/fdbserver/workloads/StatusWorkload.cpp
+++ b/fdbserver/workloads/StatusWorkload.cpp
@@ -24,7 +24,7 @@
 #include "fdbclient/StatusClient.h"
 #include "flow/UnitTest.h"
 #include "fdbclient/Schemas.h"
-#include "fdbclient/ManagementAPI.h"
+#include "fdbclient/StatusSchema.h"
 
 struct StatusWorkload : TestWorkload {
 	static constexpr auto NAME = "Status";
diff --git a/fdbserver/workloads/Throttling.cpp b/fdbserver/workloads/Throttling.cpp
index 6aa14a91d3b..2afe7b673cd 100644
--- a/fdbserver/workloads/Throttling.cpp
+++ b/fdbserver/workloads/Throttling.cpp
@@ -20,9 +20,9 @@
 
 #include <boost/lexical_cast.hpp>
 
-#include "fdbclient/ManagementAPI.h"
 #include "fdbclient/ReadYourWrites.h"
 #include "fdbclient/Schemas.h"
+#include "fdbclient/StatusSchema.h"
 #include "fdbserver/tester/workloads.h"
 
 struct TokenBucket {