From 3023096962e424b9be61993c1b6034a1fb4b8916 Mon Sep 17 00:00:00 2001
From: Junhyun Shim <junhyun.shim@snowflake.com>
Date: Thu, 8 Sep 2022 13:37:06 +0200
Subject: [PATCH 1/6] Add a knob to allow token-less tenant data access for
 untrusted clients

---
 fdbrpc/include/fdbrpc/TenantInfo.h            |  5 +--
 fdbserver/KnobProtectiveGroups.cpp            |  6 +++-
 .../fdbserver/workloads/BulkSetup.actor.h     |  4 +--
 fdbserver/tester.actor.cpp                    |  3 ++
 fdbserver/workloads/Cycle.actor.cpp           | 14 ++++----
 flow/Knobs.cpp                                |  1 +
 flow/include/flow/Knobs.h                     |  1 +
 tests/CMakeLists.txt                          |  1 +
 tests/fast/TenantCycleTokenless.toml          | 36 +++++++++++++++++++
 9 files changed, 58 insertions(+), 13 deletions(-)
 create mode 100644 tests/fast/TenantCycleTokenless.toml

diff --git a/fdbrpc/include/fdbrpc/TenantInfo.h b/fdbrpc/include/fdbrpc/TenantInfo.h
index 8d3ea46cb9..f05412d7d4 100644
--- a/fdbrpc/include/fdbrpc/TenantInfo.h
+++ b/fdbrpc/include/fdbrpc/TenantInfo.h
@@ -26,6 +26,7 @@
 #include "fdbrpc/TokenCache.h"
 #include "fdbrpc/FlowTransport.h"
 #include "flow/Arena.h"
+#include "flow/Knobs.h"
 
 struct TenantInfo {
 	static constexpr const int64_t INVALID_TENANT = -1;
@@ -69,8 +70,8 @@ struct serializable_traits<TenantInfo> : std::true_type {
 	static void serialize(Archiver& ar, TenantInfo& v) {
 		serializer(ar, v.name, v.tenantId, v.token, v.arena);
 		if constexpr (Archiver::isDeserializing) {
-			bool tenantAuthorized = false;
-			if (v.name.present() && v.token.present()) {
+			bool tenantAuthorized = FLOW_KNOBS->ALLOW_TOKENLESS_TENANT_ACCESS;
+			if (!tenantAuthorized && v.name.present() && v.token.present()) {
 				tenantAuthorized = TokenCache::instance().validate(v.name.get(), v.token.get());
 			}
 			v.trusted = FlowTransport::transport().currentDeliveryPeerIsTrusted();
diff --git a/fdbserver/KnobProtectiveGroups.cpp b/fdbserver/KnobProtectiveGroups.cpp
index 2f251035fe..a460f0966d 100644
--- a/fdbserver/KnobProtectiveGroups.cpp
+++ b/fdbserver/KnobProtectiveGroups.cpp
@@ -22,6 +22,7 @@
 
 #include <array>
 
+#include "flow/Knobs.h"
 #include "fdbclient/Knobs.h"
 #include "fdbclient/ServerKnobCollection.h"
 #include "fdbserver/Knobs.h"
@@ -52,6 +53,9 @@ void KnobProtectiveGroup::snapshotOriginalKnobs() {
 		if (std::get_if<NoKnobFound>(&value)) {
 			value = SERVER_KNOBS->getKnob(name);
 		}
+		if (std::get_if<NoKnobFound>(&value)) {
+			value = FLOW_KNOBS->getKnob(name);
+		}
 		if (std::get_if<NoKnobFound>(&value)) {
 			ASSERT(false);
 		}
@@ -70,4 +74,4 @@ void KnobProtectiveGroup::assignKnobs(const KnobKeyValuePairs& overrideKnobs) {
 		ASSERT(mutableServerKnobs.trySetKnob(name, valueRef));
 		TraceEvent(SevInfo, "AssignKnobValue").detail("KnobName", name).detail("KnobValue", valueRef.toString());
 	}
-}
\ No newline at end of file
+}
diff --git a/fdbserver/include/fdbserver/workloads/BulkSetup.actor.h b/fdbserver/include/fdbserver/workloads/BulkSetup.actor.h
index 575c361be4..4ce28a8b1b 100644
--- a/fdbserver/include/fdbserver/workloads/BulkSetup.actor.h
+++ b/fdbserver/include/fdbserver/workloads/BulkSetup.actor.h
@@ -43,7 +43,7 @@ template <class T>
 struct sfinae_true : std::true_type {};
 
 template <class T>
-auto testAuthToken(int) -> sfinae_true<decltype(std::declval<T>().getAuthToken())>;
+auto testAuthToken(int) -> sfinae_true<decltype(std::declval<T>().setAuthToken(std::declval<Transaction&>()))>;
 template <class>
 auto testAuthToken(long) -> std::false_type;
 
@@ -53,7 +53,7 @@ struct hasAuthToken : decltype(testAuthToken<T>(0)) {};
 template <class T>
 void setAuthToken(T const& self, Transaction& tr) {
 	if constexpr (hasAuthToken<T>::value) {
-		tr.setOption(FDBTransactionOptions::AUTHORIZATION_TOKEN, self.getAuthToken());
+		self.setAuthToken(tr);
 	}
 }
 
diff --git a/fdbserver/tester.actor.cpp b/fdbserver/tester.actor.cpp
index 4e3af4345e..f4158f41e2 100644
--- a/fdbserver/tester.actor.cpp
+++ b/fdbserver/tester.actor.cpp
@@ -1418,6 +1418,9 @@ KnobKeyValuePairs getOverriddenKnobKeyValues(const toml::value& context) {
 				if (std::get_if<NoKnobFound>(&parsedValue)) {
 					parsedValue = SERVER_KNOBS->parseKnobValue(key, value);
 				}
+				if (std::get_if<NoKnobFound>(&parsedValue)) {
+					parsedValue = FLOW_KNOBS->parseKnobValue(key, value);
+				}
 				if (std::get_if<NoKnobFound>(&parsedValue)) {
 					TraceEvent(SevError, "TestSpecUnrecognizedKnob")
 					    .detail("KnobName", key)
diff --git a/fdbserver/workloads/Cycle.actor.cpp b/fdbserver/workloads/Cycle.actor.cpp
index 2299367984..f6834d1fe1 100644
--- a/fdbserver/workloads/Cycle.actor.cpp
+++ b/fdbserver/workloads/Cycle.actor.cpp
@@ -43,6 +43,7 @@ struct CycleMembers<true> {
 	TenantName tenant;
 	authz::jwt::TokenRef token;
 	StringRef signedToken;
+	bool useToken;
 };
 
 template <bool MultiTenancy>
@@ -67,6 +68,7 @@ struct CycleWorkload : TestWorkload, CycleMembers<MultiTenancy> {
 		minExpectedTransactionsPerSecond = transactionsPerSecond * getOption(options, "expectedRate"_sr, 0.7);
 		if constexpr (MultiTenancy) {
 			ASSERT(g_network->isSimulated());
+			this->useToken = getOption(options, "useToken"_sr, true);
 			auto k = g_simulator.authKeys.begin();
 			this->tenant = getOption(options, "tenant"_sr, "CycleTenant"_sr);
 			// make it comfortably longer than the timeout of the workload
@@ -85,11 +87,6 @@ struct CycleWorkload : TestWorkload, CycleMembers<MultiTenancy> {
 		}
 	}
 
-	template <bool MT = MultiTenancy>
-	std::enable_if_t<MT, StringRef> getAuthToken() const {
-		return this->signedToken;
-	}
-
 	std::string description() const override {
 		if constexpr (MultiTenancy) {
 			return "TenantCycleWorkload";
@@ -151,12 +148,13 @@ struct CycleWorkload : TestWorkload, CycleMembers<MultiTenancy> {
 	}
 
 	template <bool B = MultiTenancy>
-	std::enable_if_t<B> setAuthToken(Transaction& tr) {
-		tr.setOption(FDBTransactionOptions::AUTHORIZATION_TOKEN, this->signedToken);
+	std::enable_if_t<B> setAuthToken(Transaction& tr) const {
+		if (this->useToken)
+			tr.setOption(FDBTransactionOptions::AUTHORIZATION_TOKEN, this->signedToken);
 	}
 
 	template <bool B = MultiTenancy>
-	std::enable_if_t<!B> setAuthToken(Transaction& tr) {}
+	std::enable_if_t<!B> setAuthToken(Transaction& tr) const {}
 
 	ACTOR Future<Void> cycleClient(Database cx, CycleWorkload* self, double delay) {
 		state double lastTime = now();
diff --git a/flow/Knobs.cpp b/flow/Knobs.cpp
index 916c7ec1c0..68279c26ba 100644
--- a/flow/Knobs.cpp
+++ b/flow/Knobs.cpp
@@ -131,6 +131,7 @@ void FlowKnobs::initialize(Randomize randomize, IsSimulated isSimulated) {
 	init( NETWORK_TEST_SCRIPT_MODE,                          false );
 
 	//Authorization
+	init( ALLOW_TOKENLESS_TENANT_ACCESS,                     false );
 	init( PUBLIC_KEY_FILE_MAX_SIZE,                    1024 * 1024 );
 	init( PUBLIC_KEY_FILE_REFRESH_INTERVAL_SECONDS,             30 );
 	init( MAX_CACHED_EXPIRED_TOKENS,                          1024 );
diff --git a/flow/include/flow/Knobs.h b/flow/include/flow/Knobs.h
index 35f0973cb9..fee670e880 100644
--- a/flow/include/flow/Knobs.h
+++ b/flow/include/flow/Knobs.h
@@ -196,6 +196,7 @@ public:
 	bool NETWORK_TEST_SCRIPT_MODE;
 
 	// Authorization
+	bool ALLOW_TOKENLESS_TENANT_ACCESS;
 	int PUBLIC_KEY_FILE_MAX_SIZE;
 	int PUBLIC_KEY_FILE_REFRESH_INTERVAL_SECONDS;
 	int MAX_CACHED_EXPIRED_TOKENS;
diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt
index f2a5ea76e2..055916d35b 100644
--- a/tests/CMakeLists.txt
+++ b/tests/CMakeLists.txt
@@ -184,6 +184,7 @@ if(WITH_PYTHON)
   add_fdb_test(TEST_FILES fast/SystemRebootTestCycle.toml)
   add_fdb_test(TEST_FILES fast/TaskBucketCorrectness.toml)
   add_fdb_test(TEST_FILES fast/TenantCycle.toml)
+  add_fdb_test(TEST_FILES fast/TenantCycleTokenless.toml)
   add_fdb_test(TEST_FILES fast/TenantEntryCache.toml)
   add_fdb_test(TEST_FILES fast/TimeKeeperCorrectness.toml)
   add_fdb_test(TEST_FILES fast/TxnStateStoreCycleTest.toml)
diff --git a/tests/fast/TenantCycleTokenless.toml b/tests/fast/TenantCycleTokenless.toml
new file mode 100644
index 0000000000..6e192c1f7f
--- /dev/null
+++ b/tests/fast/TenantCycleTokenless.toml
@@ -0,0 +1,36 @@
+[configuration]
+allowDefaultTenant = false
+allowDisablingTenants = false
+
+[[knobs]]
+allow_tokenless_tenant_access = true
+
+[[test]]
+testTitle = 'TenantCreation'
+
+    [[test.workload]]
+    testName = 'CreateTenant'
+    name = 'First'
+
+    [[test.workload]]
+    testName = 'CreateTenant'
+    name = 'Second'
+
+[[test]]
+testTitle = 'Cycle'
+
+    [[test.workload]]
+    testName = 'TenantCycle'
+    tenant = 'First'
+    transactionsPerSecond = 250.0
+    testDuration = 10.0
+    expectedRate = 0.80
+    useToken = false
+
+    [[test.workload]]
+    testName = 'TenantCycle'
+    tenant = 'Second'
+    transactionsPerSecond = 2500.0
+    testDuration = 10.0
+    expectedRate = 0.80
+    useToken = false

From 726d5215a06e18b718cf8d090ee4addce0c7b52a Mon Sep 17 00:00:00 2001
From: "A.J. Beamon" <aj.beamon@snowflake.com>
Date: Thu, 8 Sep 2022 08:22:36 -0700
Subject: [PATCH 2/6] Remove API 720 guards for tenants (experimental feature)
 and the cluster ID special keys (#8108)

* Remove API 720 guards for tenants (experimental feature) and the cluster ID special keys (no need to guard)

* Enable the relaxed special key access in transactions that need to use special key-space APIs introduced in 7.2
---
 .../src/main/com/apple/foundationdb/FDB.java  |  5 --
 .../apple/foundationdb/TenantManagement.java  | 19 ++++++-
 bindings/python/fdb/__init__.py               |  4 +-
 bindings/python/fdb/tenant_management.py      | 12 ++++-
 documentation/sphinx/source/special-keys.rst  |  6 +--
 fdbcli/TenantCommands.actor.cpp               | 54 ++++++-------------
 fdbcli/fdbcli.actor.cpp                       | 22 ++------
 fdbcli/include/fdbcli/fdbcli.actor.h          | 10 ++--
 fdbclient/MultiVersionTransaction.actor.cpp   |  5 +-
 fdbclient/NativeAPI.actor.cpp                 | 46 +++++++---------
 fdbclient/Tenant.cpp                          | 21 +++-----
 fdbclient/TenantSpecialKeys.cpp               | 43 ---------------
 fdbclient/include/fdbclient/Tenant.h          |  2 +-
 .../fdbclient/TenantSpecialKeys.actor.h       | 33 +++++-------
 .../TenantManagementWorkload.actor.cpp        | 12 ++---
 flow/ApiVersion.h.cmake                       |  2 -
 flow/ApiVersions.cmake                        |  4 +-
 flow/ProtocolVersion.h.cmake                  |  1 +
 flow/ProtocolVersions.cmake                   |  3 +-
 19 files changed, 111 insertions(+), 193 deletions(-)
 delete mode 100644 fdbclient/TenantSpecialKeys.cpp

diff --git a/bindings/java/src/main/com/apple/foundationdb/FDB.java b/bindings/java/src/main/com/apple/foundationdb/FDB.java
index 5215d0836e..47ba2eead1 100644
--- a/bindings/java/src/main/com/apple/foundationdb/FDB.java
+++ b/bindings/java/src/main/com/apple/foundationdb/FDB.java
@@ -191,11 +191,6 @@ public class FDB {
 		Select_API_version(version);
 		singleton = new FDB(version);
 
-		if (version < 720) {
-			TenantManagement.TENANT_MAP_PREFIX = ByteArrayUtil.join(new byte[] { (byte)255, (byte)255 },
-					"/management/tenant_map/".getBytes());
-		}
-
 		return singleton;
 	}
 
diff --git a/bindings/java/src/main/com/apple/foundationdb/TenantManagement.java b/bindings/java/src/main/com/apple/foundationdb/TenantManagement.java
index 12aaf70322..a376db79a3 100644
--- a/bindings/java/src/main/com/apple/foundationdb/TenantManagement.java
+++ b/bindings/java/src/main/com/apple/foundationdb/TenantManagement.java
@@ -52,6 +52,10 @@ public class TenantManagement {
 	 * @param tenantName The name of the tenant. Can be any byte string that does not begin a 0xFF byte.
 	 */
 	public static void createTenant(Transaction tr, byte[] tenantName) {
+		if (FDB.instance().getAPIVersion() < 720) {
+			tr.options().setSpecialKeySpaceRelaxed();
+		}
+
 		tr.options().setSpecialKeySpaceEnableWrites();
 		tr.set(ByteArrayUtil.join(TENANT_MAP_PREFIX, tenantName), new byte[0]);
 	}
@@ -86,6 +90,7 @@ public class TenantManagement {
 		final AtomicBoolean checkedExistence = new AtomicBoolean(false);
 		final byte[] key = ByteArrayUtil.join(TENANT_MAP_PREFIX, tenantName);
 		return db.runAsync(tr -> {
+			tr.options().setSpecialKeySpaceRelaxed();
 			tr.options().setSpecialKeySpaceEnableWrites();
 			if(checkedExistence.get()) {
 				tr.set(key, new byte[0]);
@@ -133,6 +138,10 @@ public class TenantManagement {
 	 * @param tenantName The name of the tenant being deleted.
 	 */
 	public static void deleteTenant(Transaction tr, byte[] tenantName) {
+		if (FDB.instance().getAPIVersion() < 720) {
+			tr.options().setSpecialKeySpaceRelaxed();
+		}
+
 		tr.options().setSpecialKeySpaceEnableWrites();
 		tr.clear(ByteArrayUtil.join(TENANT_MAP_PREFIX, tenantName));
 	}
@@ -173,6 +182,7 @@ public class TenantManagement {
 		final AtomicBoolean checkedExistence = new AtomicBoolean(false);
 		final byte[] key = ByteArrayUtil.join(TENANT_MAP_PREFIX, tenantName);
 		return db.runAsync(tr -> {
+			tr.options().setSpecialKeySpaceRelaxed();
 			tr.options().setSpecialKeySpaceEnableWrites();
 			if(checkedExistence.get()) {
 				tr.clear(key);
@@ -238,7 +248,12 @@ public class TenantManagement {
 	 * and the value is the unprocessed JSON string containing the tenant's metadata
 	 */
 	public static CloseableAsyncIterator<KeyValue> listTenants(Database db, Tuple begin, Tuple end, int limit) {
-		return listTenants_internal(db.createTransaction(), begin.pack(), end.pack(), limit);
+		Transaction tr = db.createTransaction();
+		if (FDB.instance().getAPIVersion() < 720) {
+			tr.options().setSpecialKeySpaceRelaxed();
+		}
+
+		return listTenants_internal(tr, begin.pack(), end.pack(), limit);
 	}
 
 	private static CloseableAsyncIterator<KeyValue> listTenants_internal(Transaction tr, byte[] begin, byte[] end,
@@ -262,7 +277,7 @@ public class TenantManagement {
 			this.begin = ByteArrayUtil.join(TENANT_MAP_PREFIX, begin);
 			this.end = ByteArrayUtil.join(TENANT_MAP_PREFIX, end);
 
-			tr.options().setReadSystemKeys();
+			tr.options().setRawAccess();
 			tr.options().setLockAware();
 
 			firstGet = tr.getRange(this.begin, this.end, limit);
diff --git a/bindings/python/fdb/__init__.py b/bindings/python/fdb/__init__.py
index e7d1a8bc30..930ad35396 100644
--- a/bindings/python/fdb/__init__.py
+++ b/bindings/python/fdb/__init__.py
@@ -100,10 +100,8 @@ def api_version(ver):
 
     _add_symbols(fdb.impl, list)
 
-    if ver >= 710:
+    if ver >= 630:
         import fdb.tenant_management
-        if ver < 720:
-            fdb.tenant_management._tenant_map_prefix = b'\xff\xff/management/tenant_map/'
 
     if ver < 610:
         globals()["init"] = getattr(fdb.impl, "init")
diff --git a/bindings/python/fdb/tenant_management.py b/bindings/python/fdb/tenant_management.py
index 84c3a46d03..5b67b798fa 100644
--- a/bindings/python/fdb/tenant_management.py
+++ b/bindings/python/fdb/tenant_management.py
@@ -23,6 +23,7 @@
 """Documentation for this API can be found at
 https://apple.github.io/foundationdb/api-python.html"""
 
+import fdb
 from fdb import impl as _impl
 
 _tenant_map_prefix = b'\xff\xff/management/tenant/map/'
@@ -52,6 +53,9 @@ def _check_tenant_existence(tr, key, existence_check_marker, force_maybe_commite
 # If the existence_check_marker is a non-empty list, then the existence check is skipped.
 @_impl.transactional
 def _create_tenant_impl(tr, tenant_name, existence_check_marker, force_existence_check_maybe_committed=False):
+    if fdb._version < 720:
+        tr.options.set_special_key_space_relaxed()
+
     tr.options.set_special_key_space_enable_writes()
     key = b'%s%s' % (_tenant_map_prefix, tenant_name)
 
@@ -70,6 +74,9 @@ def _create_tenant_impl(tr, tenant_name, existence_check_marker, force_existence
 # If the existence_check_marker is a non-empty list, then the existence check is skipped.
 @_impl.transactional
 def _delete_tenant_impl(tr, tenant_name, existence_check_marker, force_existence_check_maybe_committed=False):
+    if fdb._version < 720:
+        tr.options.set_special_key_space_relaxed()
+
     tr.options.set_special_key_space_enable_writes()
     key = b'%s%s' % (_tenant_map_prefix, tenant_name)
 
@@ -103,7 +110,10 @@ class FDBTenantList(object):
 # JSON strings of the tenant metadata
 @_impl.transactional
 def _list_tenants_impl(tr, begin, end, limit):
-    tr.options.set_read_system_keys()
+    if fdb._version < 720:
+        tr.options.set_special_key_space_relaxed()
+
+    tr.options.set_raw_access()
     begin_key = b'%s%s' % (_tenant_map_prefix, begin)
     end_key = b'%s%s' % (_tenant_map_prefix, end)
 
diff --git a/documentation/sphinx/source/special-keys.rst b/documentation/sphinx/source/special-keys.rst
index aa5eede4af..75877d922b 100644
--- a/documentation/sphinx/source/special-keys.rst
+++ b/documentation/sphinx/source/special-keys.rst
@@ -27,13 +27,10 @@ Each special key that existed before api version 630 is its own module. These ar
 Prior to api version 630, it was also possible to read a range starting at ``\xff\xff/worker_interfaces``. This is mostly an implementation detail of fdbcli,
 but it's available in api version 630 as a module with prefix ``\xff\xff/worker_interfaces/``.
 
-Api version 630 includes two new modules:
+Api version 630 includes three new modules:
 
 #. ``\xff\xff/transaction/`` - information about the current transaction
 #. ``\xff\xff/metrics/`` - various metrics, not transactional
-
-Api version 720 includes one new module:
-
 #. ``\xff\xff/clusterId`` - returns an immutable unique ID for a cluster
 
 Transaction module
@@ -279,7 +276,6 @@ Deprecated Keys
 Listed below are the special keys that have been deprecated. Special key(s) will no longer be accessible when the client specifies an API version equal to or larger than the version where they were deprecated. Clients specifying older API versions will be able to continue using the deprecated key(s).
 
 #. ``\xff\xff/management/profiling/<client_txn_sample_rate|client_txn_size_limit>`` Deprecated as of API version 720. The corresponding functionalities are now covered by the global configuration module. For details, see :doc:`global-configuration`. Read/write. Changing these two keys will change the corresponding system keys ``\xff\x02/fdbClientInfo/<client_txn_sample_rate|client_txn_size_limit>``, respectively. The value of ``\xff\xff/management/client_txn_sample_rate`` is a literal text of ``double``, and the value of ``\xff\xff/management/client_txn_size_limit`` is a literal text of ``int64_t``. A special value ``default`` can be set to or read from these two keys, representing the client profiling is disabled. In addition, ``clear`` in this range is not allowed. For more details, see help text of ``fdbcli`` command ``profile client``.
-#. ``\xff\xff/management/tenant_map/<tenant>`` Removed as of API version 720 and renamed to ``\xff\xff/management/tenant/map/<tenant>``.
 
 Versioning
 ==========
diff --git a/fdbcli/TenantCommands.actor.cpp b/fdbcli/TenantCommands.actor.cpp
index 7648d8dbd8..e899f960db 100644
--- a/fdbcli/TenantCommands.actor.cpp
+++ b/fdbcli/TenantCommands.actor.cpp
@@ -36,24 +36,12 @@
 
 namespace fdb_cli {
 
-const KeyRangeRef tenantMapSpecialKeyRange720("\xff\xff/management/tenant/map/"_sr,
-                                              "\xff\xff/management/tenant/map0"_sr);
+const KeyRangeRef tenantMapSpecialKeyRange("\xff\xff/management/tenant/map/"_sr, "\xff\xff/management/tenant/map0"_sr);
 const KeyRangeRef tenantConfigSpecialKeyRange("\xff\xff/management/tenant/configure/"_sr,
                                               "\xff\xff/management/tenant/configure0"_sr);
 const KeyRangeRef tenantRenameSpecialKeyRange("\xff\xff/management/tenant/rename/"_sr,
                                               "\xff\xff/management/tenant/rename0"_sr);
 
-const KeyRangeRef tenantMapSpecialKeyRange710("\xff\xff/management/tenant_map/"_sr,
-                                              "\xff\xff/management/tenant_map0"_sr);
-
-KeyRangeRef const& tenantMapSpecialKeyRange(int apiVersion) {
-	if (apiVersion >= 720) {
-		return tenantMapSpecialKeyRange720;
-	} else {
-		return tenantMapSpecialKeyRange710;
-	}
-}
-
 Optional<std::map<Standalone<StringRef>, Optional<Value>>>
 parseTenantConfiguration(std::vector<StringRef> const& tokens, int startIndex, bool allowUnset) {
 	std::map<Standalone<StringRef>, Optional<Value>> configParams;
@@ -114,13 +102,13 @@ void applyConfigurationToSpecialKeys(Reference<ITransaction> tr,
 }
 
 // createtenant command
-ACTOR Future<bool> createTenantCommandActor(Reference<IDatabase> db, std::vector<StringRef> tokens, int apiVersion) {
+ACTOR Future<bool> createTenantCommandActor(Reference<IDatabase> db, std::vector<StringRef> tokens) {
 	if (tokens.size() < 2 || tokens.size() > 3) {
 		printUsage(tokens[0]);
 		return false;
 	}
 
-	state Key tenantNameKey = tenantMapSpecialKeyRange(apiVersion).begin.withSuffix(tokens[1]);
+	state Key tenantNameKey = tenantMapSpecialKeyRange.begin.withSuffix(tokens[1]);
 	state Reference<ITransaction> tr = db->createTransaction();
 	state bool doneExistenceCheck = false;
 
@@ -131,11 +119,6 @@ ACTOR Future<bool> createTenantCommandActor(Reference<IDatabase> db, std::vector
 		return false;
 	}
 
-	if (apiVersion < 720 && !configuration.get().empty()) {
-		fmt::print(stderr, "ERROR: tenants do not accept configuration options before API version 720.\n");
-		return false;
-	}
-
 	loop {
 		try {
 			tr->setOption(FDBTransactionOptions::SPECIAL_KEY_SPACE_ENABLE_WRITES);
@@ -187,13 +170,13 @@ CommandFactory createTenantFactory(
                 "that will require this tenant to be placed on the same cluster as other tenants in the same group."));
 
 // deletetenant command
-ACTOR Future<bool> deleteTenantCommandActor(Reference<IDatabase> db, std::vector<StringRef> tokens, int apiVersion) {
+ACTOR Future<bool> deleteTenantCommandActor(Reference<IDatabase> db, std::vector<StringRef> tokens) {
 	if (tokens.size() != 2) {
 		printUsage(tokens[0]);
 		return false;
 	}
 
-	state Key tenantNameKey = tenantMapSpecialKeyRange(apiVersion).begin.withSuffix(tokens[1]);
+	state Key tenantNameKey = tenantMapSpecialKeyRange.begin.withSuffix(tokens[1]);
 	state Reference<ITransaction> tr = db->createTransaction();
 	state bool doneExistenceCheck = false;
 
@@ -243,7 +226,7 @@ CommandFactory deleteTenantFactory(
         "Deletes a tenant from the cluster. Deletion will be allowed only if the specified tenant contains no data."));
 
 // listtenants command
-ACTOR Future<bool> listTenantsCommandActor(Reference<IDatabase> db, std::vector<StringRef> tokens, int apiVersion) {
+ACTOR Future<bool> listTenantsCommandActor(Reference<IDatabase> db, std::vector<StringRef> tokens) {
 	if (tokens.size() > 4) {
 		printUsage(tokens[0]);
 		return false;
@@ -271,8 +254,8 @@ ACTOR Future<bool> listTenantsCommandActor(Reference<IDatabase> db, std::vector<
 		}
 	}
 
-	state Key beginTenantKey = tenantMapSpecialKeyRange(apiVersion).begin.withSuffix(beginTenant);
-	state Key endTenantKey = tenantMapSpecialKeyRange(apiVersion).begin.withSuffix(endTenant);
+	state Key beginTenantKey = tenantMapSpecialKeyRange.begin.withSuffix(beginTenant);
+	state Key endTenantKey = tenantMapSpecialKeyRange.begin.withSuffix(endTenant);
 	state Reference<ITransaction> tr = db->createTransaction();
 
 	loop {
@@ -292,7 +275,7 @@ ACTOR Future<bool> listTenantsCommandActor(Reference<IDatabase> db, std::vector<
 				    tr->getRange(firstGreaterOrEqual(beginTenantKey), firstGreaterOrEqual(endTenantKey), limit);
 				RangeResult tenants = wait(safeThreadFutureToFuture(kvsFuture));
 				for (auto tenant : tenants) {
-					tenantNames.push_back(tenant.key.removePrefix(tenantMapSpecialKeyRange(apiVersion).begin));
+					tenantNames.push_back(tenant.key.removePrefix(tenantMapSpecialKeyRange.begin));
 				}
 			}
 
@@ -330,14 +313,14 @@ CommandFactory listTenantsFactory(
                 "The number of tenants to print can be specified using the [LIMIT] parameter, which defaults to 100."));
 
 // gettenant command
-ACTOR Future<bool> getTenantCommandActor(Reference<IDatabase> db, std::vector<StringRef> tokens, int apiVersion) {
+ACTOR Future<bool> getTenantCommandActor(Reference<IDatabase> db, std::vector<StringRef> tokens) {
 	if (tokens.size() < 2 || tokens.size() > 3 || (tokens.size() == 3 && tokens[2] != "JSON"_sr)) {
 		printUsage(tokens[0]);
 		return false;
 	}
 
 	state bool useJson = tokens.size() == 3;
-	state Key tenantNameKey = tenantMapSpecialKeyRange(apiVersion).begin.withSuffix(tokens[1]);
+	state Key tenantNameKey = tenantMapSpecialKeyRange.begin.withSuffix(tokens[1]);
 	state Reference<ITransaction> tr = db->createTransaction();
 
 	loop {
@@ -347,7 +330,7 @@ ACTOR Future<bool> getTenantCommandActor(Reference<IDatabase> db, std::vector<St
 			state std::string tenantJson;
 			if (clusterType == ClusterType::METACLUSTER_MANAGEMENT) {
 				TenantMapEntry entry = wait(MetaclusterAPI::getTenantTransaction(tr, tokens[1]));
-				tenantJson = entry.toJson(apiVersion);
+				tenantJson = entry.toJson();
 			} else {
 				// Hold the reference to the standalone's memory
 				state ThreadFuture<Optional<Value>> tenantFuture = tr->get(tenantNameKey);
@@ -378,12 +361,7 @@ ACTOR Future<bool> getTenantCommandActor(Reference<IDatabase> db, std::vector<St
 				std::string assignedCluster;
 
 				doc.get("id", id);
-
-				if (apiVersion >= 720) {
-					doc.get("prefix.printable", prefix);
-				} else {
-					doc.get("prefix", prefix);
-				}
+				doc.get("prefix.printable", prefix);
 
 				doc.get("tenant_state", tenantState);
 				bool hasTenantGroup = doc.tryGet("tenant_group.printable", tenantGroup);
@@ -499,15 +477,15 @@ int64_t getTenantId(Value metadata) {
 }
 
 // renametenant command
-ACTOR Future<bool> renameTenantCommandActor(Reference<IDatabase> db, std::vector<StringRef> tokens, int apiVersion) {
+ACTOR Future<bool> renameTenantCommandActor(Reference<IDatabase> db, std::vector<StringRef> tokens) {
 	if (tokens.size() != 3) {
 		printUsage(tokens[0]);
 		return false;
 	}
 	state Reference<ITransaction> tr = db->createTransaction();
 	state Key tenantRenameKey = tenantRenameSpecialKeyRange.begin.withSuffix(tokens[1]);
-	state Key tenantOldNameKey = tenantMapSpecialKeyRange(apiVersion).begin.withSuffix(tokens[1]);
-	state Key tenantNewNameKey = tenantMapSpecialKeyRange(apiVersion).begin.withSuffix(tokens[2]);
+	state Key tenantOldNameKey = tenantMapSpecialKeyRange.begin.withSuffix(tokens[1]);
+	state Key tenantNewNameKey = tenantMapSpecialKeyRange.begin.withSuffix(tokens[2]);
 	state bool firstTry = true;
 	state int64_t id = -1;
 	loop {
diff --git a/fdbcli/fdbcli.actor.cpp b/fdbcli/fdbcli.actor.cpp
index 12106dd9d4..ca357acde5 100644
--- a/fdbcli/fdbcli.actor.cpp
+++ b/fdbcli/fdbcli.actor.cpp
@@ -1875,14 +1875,14 @@ ACTOR Future<int> cli(CLIOptions opt, LineNoise* plinenoise, Reference<ClusterCo
 				}
 
 				if (tokencmp(tokens[0], "createtenant")) {
-					bool _result = wait(makeInterruptable(createTenantCommandActor(db, tokens, opt.apiVersion)));
+					bool _result = wait(makeInterruptable(createTenantCommandActor(db, tokens)));
 					if (!_result)
 						is_error = true;
 					continue;
 				}
 
 				if (tokencmp(tokens[0], "deletetenant")) {
-					bool _result = wait(makeInterruptable(deleteTenantCommandActor(db, tokens, opt.apiVersion)));
+					bool _result = wait(makeInterruptable(deleteTenantCommandActor(db, tokens)));
 					if (!_result)
 						is_error = true;
 					else if (tenantName.present() && tokens[1] == tenantName.get()) {
@@ -1894,26 +1894,20 @@ ACTOR Future<int> cli(CLIOptions opt, LineNoise* plinenoise, Reference<ClusterCo
 				}
 
 				if (tokencmp(tokens[0], "listtenants")) {
-					bool _result = wait(makeInterruptable(listTenantsCommandActor(db, tokens, opt.apiVersion)));
+					bool _result = wait(makeInterruptable(listTenantsCommandActor(db, tokens)));
 					if (!_result)
 						is_error = true;
 					continue;
 				}
 
 				if (tokencmp(tokens[0], "gettenant")) {
-					bool _result = wait(makeInterruptable(getTenantCommandActor(db, tokens, opt.apiVersion)));
+					bool _result = wait(makeInterruptable(getTenantCommandActor(db, tokens)));
 					if (!_result)
 						is_error = true;
 					continue;
 				}
 
 				if (tokencmp(tokens[0], "configuretenant")) {
-					if (opt.apiVersion < 720) {
-						fmt::print(stderr, "ERROR: tenants cannot be configured before API version 720.\n");
-						is_error = true;
-						continue;
-					}
-
 					bool _result = wait(makeInterruptable(configureTenantCommandActor(db, tokens)));
 					if (!_result)
 						is_error = true;
@@ -1921,13 +1915,7 @@ ACTOR Future<int> cli(CLIOptions opt, LineNoise* plinenoise, Reference<ClusterCo
 				}
 
 				if (tokencmp(tokens[0], "renametenant")) {
-					if (opt.apiVersion < 720) {
-						fmt::print(stderr, "ERROR: tenants cannot be renamed before API version 720.\n");
-						is_error = true;
-						continue;
-					}
-
-					bool _result = wait(makeInterruptable(renameTenantCommandActor(db, tokens, opt.apiVersion)));
+					bool _result = wait(makeInterruptable(renameTenantCommandActor(db, tokens)));
 					if (!_result)
 						is_error = true;
 					continue;
diff --git a/fdbcli/include/fdbcli/fdbcli.actor.h b/fdbcli/include/fdbcli/fdbcli.actor.h
index 3df51b4677..fa0c4eebe0 100644
--- a/fdbcli/include/fdbcli/fdbcli.actor.h
+++ b/fdbcli/include/fdbcli/fdbcli.actor.h
@@ -168,11 +168,11 @@ ACTOR Future<bool> consistencyCheckCommandActor(Reference<ITransaction> tr,
 // coordinators command
 ACTOR Future<bool> coordinatorsCommandActor(Reference<IDatabase> db, std::vector<StringRef> tokens);
 // createtenant command
-ACTOR Future<bool> createTenantCommandActor(Reference<IDatabase> db, std::vector<StringRef> tokens, int apiVersion);
+ACTOR Future<bool> createTenantCommandActor(Reference<IDatabase> db, std::vector<StringRef> tokens);
 // datadistribution command
 ACTOR Future<bool> dataDistributionCommandActor(Reference<IDatabase> db, std::vector<StringRef> tokens);
 // deletetenant command
-ACTOR Future<bool> deleteTenantCommandActor(Reference<IDatabase> db, std::vector<StringRef> tokens, int apiVersion);
+ACTOR Future<bool> deleteTenantCommandActor(Reference<IDatabase> db, std::vector<StringRef> tokens);
 // exclude command
 ACTOR Future<bool> excludeCommandActor(Reference<IDatabase> db, std::vector<StringRef> tokens, Future<Void> warn);
 // expensive_data_check command
@@ -189,7 +189,7 @@ ACTOR Future<bool> fileConfigureCommandActor(Reference<IDatabase> db,
 // force_recovery_with_data_loss command
 ACTOR Future<bool> forceRecoveryWithDataLossCommandActor(Reference<IDatabase> db, std::vector<StringRef> tokens);
 // gettenant command
-ACTOR Future<bool> getTenantCommandActor(Reference<IDatabase> db, std::vector<StringRef> tokens, int apiVersion);
+ACTOR Future<bool> getTenantCommandActor(Reference<IDatabase> db, std::vector<StringRef> tokens);
 // include command
 ACTOR Future<bool> includeCommandActor(Reference<IDatabase> db, std::vector<StringRef> tokens);
 // kill command
@@ -198,7 +198,7 @@ ACTOR Future<bool> killCommandActor(Reference<IDatabase> db,
                                     std::vector<StringRef> tokens,
                                     std::map<Key, std::pair<Value, ClientLeaderRegInterface>>* address_interface);
 // listtenants command
-ACTOR Future<bool> listTenantsCommandActor(Reference<IDatabase> db, std::vector<StringRef> tokens, int apiVersion);
+ACTOR Future<bool> listTenantsCommandActor(Reference<IDatabase> db, std::vector<StringRef> tokens);
 // lock/unlock command
 ACTOR Future<bool> lockCommandActor(Reference<IDatabase> db, std::vector<StringRef> tokens);
 ACTOR Future<bool> unlockDatabaseActor(Reference<IDatabase> db, UID uid);
@@ -227,7 +227,7 @@ ACTOR Future<bool> profileCommandActor(Database db,
                                        std::vector<StringRef> tokens,
                                        bool intrans);
 // renametenant command
-ACTOR Future<bool> renameTenantCommandActor(Reference<IDatabase> db, std::vector<StringRef> tokens, int apiVersion);
+ACTOR Future<bool> renameTenantCommandActor(Reference<IDatabase> db, std::vector<StringRef> tokens);
 // quota command
 ACTOR Future<bool> quotaCommandActor(Reference<IDatabase> db, std::vector<StringRef> tokens);
 // setclass command
diff --git a/fdbclient/MultiVersionTransaction.actor.cpp b/fdbclient/MultiVersionTransaction.actor.cpp
index 40dd9fbc0b..f84bcdf974 100644
--- a/fdbclient/MultiVersionTransaction.actor.cpp
+++ b/fdbclient/MultiVersionTransaction.actor.cpp
@@ -2756,12 +2756,13 @@ ACTOR Future<std::string> updateClusterSharedStateMapImpl(MultiVersionApi* self,
                                                           ProtocolVersion dbProtocolVersion,
                                                           Reference<IDatabase> db) {
 	// The cluster ID will be the connection record string (either a filename or the connection string itself)
-	// in API versions before we could read the cluster ID.
+	// in versions before we could read the cluster ID.
 	state std::string clusterId = connectionRecord.toString();
-	if (MultiVersionApi::api->getApiVersion().hasCreateDBFromConnString()) {
+	if (dbProtocolVersion.hasClusterIdSpecialKey()) {
 		state Reference<ITransaction> tr = db->createTransaction();
 		loop {
 			try {
+				tr->setOption(FDBTransactionOptions::SPECIAL_KEY_SPACE_RELAXED);
 				state ThreadFuture<Optional<Value>> clusterIdFuture = tr->get("\xff\xff/cluster_id"_sr);
 				Optional<Value> clusterIdVal = wait(safeThreadFutureToFuture(clusterIdFuture));
 				ASSERT(clusterIdVal.present());
diff --git a/fdbclient/NativeAPI.actor.cpp b/fdbclient/NativeAPI.actor.cpp
index b9d5c6237b..bfd5ab105e 100644
--- a/fdbclient/NativeAPI.actor.cpp
+++ b/fdbclient/NativeAPI.actor.cpp
@@ -1501,32 +1501,6 @@ DatabaseContext::DatabaseContext(Reference<AsyncVar<Reference<IClusterConnection
 	smoothMidShardSize.reset(CLIENT_KNOBS->INIT_MID_SHARD_BYTES);
 	globalConfig = std::make_unique<GlobalConfig>(this);
 
-	if (apiVersion.hasTenantsV2()) {
-		registerSpecialKeysImpl(
-		    SpecialKeySpace::MODULE::CLUSTERID,
-		    SpecialKeySpace::IMPLTYPE::READONLY,
-		    std::make_unique<SingleSpecialKeyImpl>(
-		        LiteralStringRef("\xff\xff/cluster_id"), [](ReadYourWritesTransaction* ryw) -> Future<Optional<Value>> {
-			        try {
-				        if (ryw->getDatabase().getPtr()) {
-					        return map(getClusterId(ryw->getDatabase()),
-					                   [](UID id) { return Optional<Value>(StringRef(id.toString())); });
-				        }
-			        } catch (Error& e) {
-				        return e;
-			        }
-			        return Optional<Value>();
-		        }));
-		registerSpecialKeysImpl(
-		    SpecialKeySpace::MODULE::MANAGEMENT,
-		    SpecialKeySpace::IMPLTYPE::READWRITE,
-		    std::make_unique<TenantRangeImpl<true>>(SpecialKeySpace::getManagementApiCommandRange("tenant")));
-	} else if (apiVersion.hasTenantsV1()) {
-		registerSpecialKeysImpl(
-		    SpecialKeySpace::MODULE::MANAGEMENT,
-		    SpecialKeySpace::IMPLTYPE::READWRITE,
-		    std::make_unique<TenantRangeImpl<false>>(SpecialKeySpace::getManagementApiCommandRange("tenantmap")));
-	}
 	if (apiVersion.version() >= 700) {
 		registerSpecialKeysImpl(SpecialKeySpace::MODULE::ERRORMSG,
 		                        SpecialKeySpace::IMPLTYPE::READONLY,
@@ -1720,6 +1694,26 @@ DatabaseContext::DatabaseContext(Reference<AsyncVar<Reference<IClusterConnection
 			        }
 			        return Optional<Value>();
 		        }));
+		registerSpecialKeysImpl(
+		    SpecialKeySpace::MODULE::CLUSTERID,
+		    SpecialKeySpace::IMPLTYPE::READONLY,
+		    std::make_unique<SingleSpecialKeyImpl>(
+		        LiteralStringRef("\xff\xff/cluster_id"), [](ReadYourWritesTransaction* ryw) -> Future<Optional<Value>> {
+			        try {
+				        if (ryw->getDatabase().getPtr()) {
+					        return map(getClusterId(ryw->getDatabase()),
+					                   [](UID id) { return Optional<Value>(StringRef(id.toString())); });
+				        }
+			        } catch (Error& e) {
+				        return e;
+			        }
+			        return Optional<Value>();
+		        }));
+
+		registerSpecialKeysImpl(
+		    SpecialKeySpace::MODULE::MANAGEMENT,
+		    SpecialKeySpace::IMPLTYPE::READWRITE,
+		    std::make_unique<TenantRangeImpl>(SpecialKeySpace::getManagementApiCommandRange("tenant")));
 	}
 	throttleExpirer = recurring([this]() { expireThrottles(); }, CLIENT_KNOBS->TAG_THROTTLE_EXPIRATION_INTERVAL);
 
diff --git a/fdbclient/Tenant.cpp b/fdbclient/Tenant.cpp
index 7445252670..835dcb32b9 100644
--- a/fdbclient/Tenant.cpp
+++ b/fdbclient/Tenant.cpp
@@ -123,24 +123,19 @@ void TenantMapEntry::setId(int64_t id) {
 	prefix = idToPrefix(id);
 }
 
-std::string TenantMapEntry::toJson(int apiVersion) const {
+std::string TenantMapEntry::toJson() const {
 	json_spirit::mObject tenantEntry;
 	tenantEntry["id"] = id;
 	tenantEntry["encrypted"] = encrypted;
 
-	if (apiVersion >= ApiVersion::withTenantsV2().version()) {
-		json_spirit::mObject prefixObject;
-		std::string encodedPrefix = base64::encoder::from_string(prefix.toString());
-		// Remove trailing newline
-		encodedPrefix.resize(encodedPrefix.size() - 1);
+	json_spirit::mObject prefixObject;
+	std::string encodedPrefix = base64::encoder::from_string(prefix.toString());
+	// Remove trailing newline
+	encodedPrefix.resize(encodedPrefix.size() - 1);
 
-		prefixObject["base64"] = encodedPrefix;
-		prefixObject["printable"] = printable(prefix);
-		tenantEntry["prefix"] = prefixObject;
-	} else {
-		// This is not a standard encoding in JSON, and some libraries may not be able to easily decode it
-		tenantEntry["prefix"] = prefix.toString();
-	}
+	prefixObject["base64"] = encodedPrefix;
+	prefixObject["printable"] = printable(prefix);
+	tenantEntry["prefix"] = prefixObject;
 
 	tenantEntry["tenant_state"] = TenantMapEntry::tenantStateToString(tenantState);
 	if (assignedCluster.present()) {
diff --git a/fdbclient/TenantSpecialKeys.cpp b/fdbclient/TenantSpecialKeys.cpp
deleted file mode 100644
index 5570816684..0000000000
--- a/fdbclient/TenantSpecialKeys.cpp
+++ /dev/null
@@ -1,43 +0,0 @@
-/*
- * TenantSpecialKeys.cpp
- *
- * This source file is part of the FoundationDB open source project
- *
- * Copyright 2013-2022 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/TenantSpecialKeys.actor.h"
-
-template <>
-const KeyRangeRef TenantRangeImpl<true>::submoduleRange = KeyRangeRef("tenant/"_sr, "tenant0"_sr);
-
-template <>
-const KeyRangeRef TenantRangeImpl<true>::mapSubRange = KeyRangeRef("map/"_sr, "map0"_sr);
-
-template <>
-const KeyRangeRef TenantRangeImpl<false>::submoduleRange = KeyRangeRef(""_sr, "\xff"_sr);
-
-template <>
-const KeyRangeRef TenantRangeImpl<false>::mapSubRange = KeyRangeRef("tenant_map/"_sr, "tenant_map0"_sr);
-
-template <>
-bool TenantRangeImpl<true>::subRangeIntersects(KeyRangeRef subRange, KeyRangeRef range) {
-	return subRange.intersects(range);
-}
-
-template <>
-bool TenantRangeImpl<false>::subRangeIntersects(KeyRangeRef subRange, KeyRangeRef range) {
-	return subRange == mapSubRange;
-}
\ No newline at end of file
diff --git a/fdbclient/include/fdbclient/Tenant.h b/fdbclient/include/fdbclient/Tenant.h
index 0af19b85f1..d75629da10 100644
--- a/fdbclient/include/fdbclient/Tenant.h
+++ b/fdbclient/include/fdbclient/Tenant.h
@@ -96,7 +96,7 @@ struct TenantMapEntry {
 	TenantMapEntry(int64_t id, TenantState tenantState, Optional<TenantGroupName> tenantGroup, bool encrypted);
 
 	void setId(int64_t id);
-	std::string toJson(int apiVersion) const;
+	std::string toJson() const;
 
 	bool matchesConfiguration(TenantMapEntry const& other) const;
 	void configure(Standalone<StringRef> parameter, Optional<Value> value);
diff --git a/fdbclient/include/fdbclient/TenantSpecialKeys.actor.h b/fdbclient/include/fdbclient/TenantSpecialKeys.actor.h
index 0613c411d4..cf190fc77d 100644
--- a/fdbclient/include/fdbclient/TenantSpecialKeys.actor.h
+++ b/fdbclient/include/fdbclient/TenantSpecialKeys.actor.h
@@ -36,11 +36,8 @@
 #include "flow/UnitTest.h"
 #include "flow/actorcompiler.h" // This must be the last #include.
 
-template <bool HasSubRanges>
 class TenantRangeImpl : public SpecialKeyRangeRWImpl {
 private:
-	static bool subRangeIntersects(KeyRangeRef subRange, KeyRangeRef range);
-
 	static KeyRangeRef removePrefix(KeyRangeRef range, KeyRef prefix, KeyRef defaultEnd) {
 		KeyRef begin = range.begin.removePrefix(prefix);
 		KeyRef end;
@@ -76,7 +73,7 @@ private:
 		    wait(TenantAPI::listTenantsTransaction(&ryw->getTransaction(), kr.begin, kr.end, limitsHint.rows));
 
 		for (auto tenant : tenants) {
-			std::string jsonString = tenant.second.toJson(ryw->getDatabase()->apiVersion.version());
+			std::string jsonString = tenant.second.toJson();
 			ValueRef tenantEntryBytes(results->arena(), jsonString);
 			results->push_back(results->arena(),
 			                   KeyValueRef(withTenantMapPrefix(tenant.first, results->arena()), tenantEntryBytes));
@@ -85,21 +82,20 @@ private:
 		return Void();
 	}
 
-	ACTOR template <bool B>
-	static Future<RangeResult> getTenantRange(ReadYourWritesTransaction* ryw,
-	                                          KeyRangeRef kr,
-	                                          GetRangeLimits limitsHint) {
+	ACTOR static Future<RangeResult> getTenantRange(ReadYourWritesTransaction* ryw,
+	                                                KeyRangeRef kr,
+	                                                GetRangeLimits limitsHint) {
 		state RangeResult results;
 
 		kr = kr.removePrefix(SpecialKeySpace::getModuleRange(SpecialKeySpace::MODULE::MANAGEMENT).begin)
-		         .removePrefix(TenantRangeImpl<B>::submoduleRange.begin);
+		         .removePrefix(TenantRangeImpl::submoduleRange.begin);
 
-		if (kr.intersects(TenantRangeImpl<B>::mapSubRange)) {
+		if (kr.intersects(TenantRangeImpl::mapSubRange)) {
 			GetRangeLimits limits = limitsHint;
 			limits.decrement(results);
 			wait(getTenantList(
 			    ryw,
-			    removePrefix(kr & TenantRangeImpl<B>::mapSubRange, TenantRangeImpl<B>::mapSubRange.begin, "\xff"_sr),
+			    removePrefix(kr & TenantRangeImpl::mapSubRange, TenantRangeImpl::mapSubRange.begin, "\xff"_sr),
 			    &results,
 			    limits));
 		}
@@ -254,11 +250,8 @@ private:
 	}
 
 public:
-	// These ranges vary based on the template parameter
-	const static KeyRangeRef submoduleRange;
-	const static KeyRangeRef mapSubRange;
-
-	// These sub-ranges should only be used if HasSubRanges=true
+	const inline static KeyRangeRef submoduleRange = KeyRangeRef("tenant/"_sr, "tenant0"_sr);
+	const inline static KeyRangeRef mapSubRange = KeyRangeRef("map/"_sr, "map0"_sr);
 	const inline static KeyRangeRef configureSubRange = KeyRangeRef("configure/"_sr, "configure0"_sr);
 	const inline static KeyRangeRef renameSubRange = KeyRangeRef("rename/"_sr, "rename0"_sr);
 
@@ -267,7 +260,7 @@ public:
 	Future<RangeResult> getRange(ReadYourWritesTransaction* ryw,
 	                             KeyRangeRef kr,
 	                             GetRangeLimits limitsHint) const override {
-		return getTenantRange<HasSubRanges>(ryw, kr, limitsHint);
+		return getTenantRange(ryw, kr, limitsHint);
 	}
 
 	ACTOR static Future<Optional<std::string>> commitImpl(TenantRangeImpl* self, ReadYourWritesTransaction* ryw) {
@@ -301,11 +294,11 @@ public:
 			        .removePrefix(SpecialKeySpace::getModuleRange(SpecialKeySpace::MODULE::MANAGEMENT).begin)
 			        .removePrefix(submoduleRange.begin);
 
-			if (subRangeIntersects(mapSubRange, adjustedRange)) {
+			if (mapSubRange.intersects(adjustedRange)) {
 				adjustedRange = mapSubRange & adjustedRange;
 				adjustedRange = removePrefix(adjustedRange, mapSubRange.begin, "\xff"_sr);
 				mapMutations.push_back(std::make_pair(adjustedRange, range.value().second));
-			} else if (subRangeIntersects(configureSubRange, adjustedRange) && adjustedRange.singleKeyRange()) {
+			} else if (configureSubRange.intersects(adjustedRange) && adjustedRange.singleKeyRange()) {
 				StringRef configTupleStr = adjustedRange.begin.removePrefix(configureSubRange.begin);
 				try {
 					Tuple tuple = Tuple::unpack(configTupleStr);
@@ -320,7 +313,7 @@ public:
 					    false, "configure tenant", "invalid tenant configuration key"));
 					throw special_keys_api_failure();
 				}
-			} else if (subRangeIntersects(renameSubRange, adjustedRange)) {
+			} else if (renameSubRange.intersects(adjustedRange)) {
 				StringRef oldName = adjustedRange.begin.removePrefix(renameSubRange.begin);
 				StringRef newName = range.value().second.get();
 				// Do not allow overlapping renames in the same commit
diff --git a/fdbserver/workloads/TenantManagementWorkload.actor.cpp b/fdbserver/workloads/TenantManagementWorkload.actor.cpp
index 6c1006d30e..3ecdf4d776 100644
--- a/fdbserver/workloads/TenantManagementWorkload.actor.cpp
+++ b/fdbserver/workloads/TenantManagementWorkload.actor.cpp
@@ -73,14 +73,14 @@ struct TenantManagementWorkload : TestWorkload {
 	TenantName localTenantGroupNamePrefix;
 
 	const Key specialKeysTenantMapPrefix = SpecialKeySpace::getModuleRange(SpecialKeySpace::MODULE::MANAGEMENT)
-	                                           .begin.withSuffix(TenantRangeImpl<true>::submoduleRange.begin)
-	                                           .withSuffix(TenantRangeImpl<true>::mapSubRange.begin);
+	                                           .begin.withSuffix(TenantRangeImpl::submoduleRange.begin)
+	                                           .withSuffix(TenantRangeImpl::mapSubRange.begin);
 	const Key specialKeysTenantConfigPrefix = SpecialKeySpace::getModuleRange(SpecialKeySpace::MODULE::MANAGEMENT)
-	                                              .begin.withSuffix(TenantRangeImpl<true>::submoduleRange.begin)
-	                                              .withSuffix(TenantRangeImpl<true>::configureSubRange.begin);
+	                                              .begin.withSuffix(TenantRangeImpl::submoduleRange.begin)
+	                                              .withSuffix(TenantRangeImpl::configureSubRange.begin);
 	const Key specialKeysTenantRenamePrefix = SpecialKeySpace::getModuleRange(SpecialKeySpace::MODULE::MANAGEMENT)
-	                                              .begin.withSuffix(TenantRangeImpl<true>::submoduleRange.begin)
-	                                              .withSuffix(TenantRangeImpl<true>::renameSubRange.begin);
+	                                              .begin.withSuffix(TenantRangeImpl::submoduleRange.begin)
+	                                              .withSuffix(TenantRangeImpl::renameSubRange.begin);
 
 	int maxTenants;
 	int maxTenantGroups;
diff --git a/flow/ApiVersion.h.cmake b/flow/ApiVersion.h.cmake
index 54350c152a..780f694907 100644
--- a/flow/ApiVersion.h.cmake
+++ b/flow/ApiVersion.h.cmake
@@ -66,12 +66,10 @@ public: // introduced features
     API_VERSION_FEATURE(@FDB_AV_PERSISTENT_OPTIONS@, PersistentOptions);
     API_VERSION_FEATURE(@FDB_AV_TRACE_FILE_IDENTIFIER@, TraceFileIdentifier);
     API_VERSION_FEATURE(@FDB_AV_CLUSTER_SHARED_STATE_MAP@, ClusterSharedStateMap);
-    API_VERSION_FEATURE(@FDB_AV_TENANTS_V1@, TenantsV1);
     API_VERSION_FEATURE(@FDB_AV_BLOB_RANGE_API@, BlobRangeApi);
     API_VERSION_FEATURE(@FDB_AV_CREATE_DB_FROM_CONN_STRING@, CreateDBFromConnString);
     API_VERSION_FEATURE(@FDB_AV_FUTURE_GET_BOOL@, FutureGetBool);
     API_VERSION_FEATURE(@FDB_AV_FUTURE_PROTOCOL_VERSION_API@, FutureProtocolVersionApi);
-    API_VERSION_FEATURE(@FDB_AV_TENANTS_V2@, TenantsV2);
 };
 
 #endif // FLOW_CODE_API_VERSION_H
diff --git a/flow/ApiVersions.cmake b/flow/ApiVersions.cmake
index 09236dace8..ad7ef59c35 100644
--- a/flow/ApiVersions.cmake
+++ b/flow/ApiVersions.cmake
@@ -7,9 +7,7 @@ set(FDB_AV_INLINE_UPDATE_DATABASE           "610")
 set(FDB_AV_PERSISTENT_OPTIONS               "610")
 set(FDB_AV_TRACE_FILE_IDENTIFIER            "630")
 set(FDB_AV_CLUSTER_SHARED_STATE_MAP         "710")
-set(FDB_AV_TENANTS_V1                       "720")
 set(FDB_AV_BLOB_RANGE_API                   "720")
 set(FDB_AV_CREATE_DB_FROM_CONN_STRING       "720")
 set(FDB_AV_FUTURE_GET_BOOL                  "720")
-set(FDB_AV_FUTURE_PROTOCOL_VERSION_API      "720")
-set(FDB_AV_TENANTS_V2                       "720")
\ No newline at end of file
+set(FDB_AV_FUTURE_PROTOCOL_VERSION_API      "720")
\ No newline at end of file
diff --git a/flow/ProtocolVersion.h.cmake b/flow/ProtocolVersion.h.cmake
index 6c33915d83..129919a091 100644
--- a/flow/ProtocolVersion.h.cmake
+++ b/flow/ProtocolVersion.h.cmake
@@ -172,6 +172,7 @@ public: // introduced features
 	PROTOCOL_VERSION_FEATURE(@FDB_PV_SHARD_ENCODE_LOCATION_METADATA@, ShardEncodeLocationMetaData);
 	PROTOCOL_VERSION_FEATURE(@FDB_PV_TENANTS@, Tenants);
 	PROTOCOL_VERSION_FEATURE(@FDB_PV_BLOB_GRANULE_FILE@, BlobGranuleFile);
+	PROTOCOL_VERSION_FEATURE(@FDB_PV_CLUSTER_ID_SPECIAL_KEY@, ClusterIdSpecialKey);
 };
 
 template <>
diff --git a/flow/ProtocolVersions.cmake b/flow/ProtocolVersions.cmake
index 92f455ae2a..b66ca6f674 100644
--- a/flow/ProtocolVersions.cmake
+++ b/flow/ProtocolVersions.cmake
@@ -87,4 +87,5 @@ set(FDB_PV_SW_VERSION_TRACKING                  "0x0FDB00B072000000LL")
 set(FDB_PV_ENCRYPTION_AT_REST                   "0x0FDB00B072000000LL")
 set(FDB_PV_SHARD_ENCODE_LOCATION_METADATA       "0x0FDB00B072000000LL")
 set(FDB_PV_TENANTS                              "0x0FDB00B072000000LL")
-set(FDB_PV_BLOB_GRANULE_FILE                    "0x0FDB00B072000000LL")
\ No newline at end of file
+set(FDB_PV_BLOB_GRANULE_FILE                    "0x0FDB00B072000000LL")
+set(FDB_PV_CLUSTER_ID_SPECIAL_KEY               "0x0FDB00B072000000LL")
\ No newline at end of file

From 984150bf74c3dd00eaebe0b33ef3e39e23b7510d Mon Sep 17 00:00:00 2001
From: Andrew Noyes <andrew.noyes@snowflake.com>
Date: Thu, 8 Sep 2022 08:27:57 -0700
Subject: [PATCH 3/6] Link fdbcli with -rdynamic for ubsan builds (#8033)

Closes #7957
---
 fdbcli/CMakeLists.txt | 7 +++++++
 1 file changed, 7 insertions(+)

diff --git a/fdbcli/CMakeLists.txt b/fdbcli/CMakeLists.txt
index aff53b1f63..d2966df807 100644
--- a/fdbcli/CMakeLists.txt
+++ b/fdbcli/CMakeLists.txt
@@ -3,6 +3,13 @@ fdb_find_sources(FDBCLI_SRCS)
 add_flow_target(EXECUTABLE NAME fdbcli SRCS ${FDBCLI_SRCS})
 target_include_directories(fdbcli PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/include" "${CMAKE_CURRENT_BINARY_DIR}/include")
 target_link_libraries(fdbcli PRIVATE fdbclient SimpleOpt)
+if (USE_UBSAN)
+  # The intent is to put typeinfo symbols in the dynamic symbol table so that
+  # the types in fdbcli and external libfdb_c clients agree for ubsan's vptr
+  # check. This would not be a good idea for the normal build, or if we ever
+  # start testing old libfdb_c's that are ubsan-instrumented.
+  target_link_options(fdbcli PRIVATE "-rdynamic")
+endif()
 
 if(NOT WIN32)
   target_link_libraries(fdbcli PRIVATE linenoise)

From bbc30c6a80d1c847b29765f7300b1209f3c3c439 Mon Sep 17 00:00:00 2001
From: Andrew Noyes <andrew.noyes@snowflake.com>
Date: Thu, 8 Sep 2022 08:31:14 -0700
Subject: [PATCH 4/6] Assert that Arena's appear last in serializer calls
 (#8078)

* Assert that arena's appear last in serializer calls

* Fix all occurrences of Arena's not appearing last in serializer call

* Work around issue from Standalone inheriting from Arena privately

* Attempt to fix windows build

Use fb_ prefix instead of detail namespace to scope implementation
details in headers
---
 .../include/fdbclient/CommitProxyInterface.h  |  6 +--
 .../fdbclient/EncryptKeyProxyInterface.h      |  2 +-
 .../fdbclient/StorageServerInterface.h        | 18 ++++-----
 .../fdbserver/DataDistributorInterface.h      |  2 +-
 .../include/fdbserver/ResolverInterface.h     |  4 +-
 fdbserver/include/fdbserver/TLogInterface.h   |  4 +-
 flow/flat_buffers.cpp                         |  6 +--
 flow/include/flow/Arena.h                     |  3 ++
 flow/include/flow/ObjectSerializerTraits.h    | 39 ++++++++++++++++---
 flow/include/flow/serialize.h                 |  4 ++
 10 files changed, 62 insertions(+), 26 deletions(-)

diff --git a/fdbclient/include/fdbclient/CommitProxyInterface.h b/fdbclient/include/fdbclient/CommitProxyInterface.h
index 1a6a0410ae..7fc516dcdd 100644
--- a/fdbclient/include/fdbclient/CommitProxyInterface.h
+++ b/fdbclient/include/fdbclient/CommitProxyInterface.h
@@ -197,7 +197,7 @@ struct CommitTransactionRequest : TimedRequest {
 	template <class Ar>
 	void serialize(Ar& ar) {
 		serializer(
-		    ar, transaction, reply, arena, flags, debugID, commitCostEstimation, tagSet, spanContext, tenantInfo);
+		    ar, transaction, reply, flags, debugID, commitCostEstimation, tagSet, spanContext, tenantInfo, arena);
 	}
 };
 
@@ -339,7 +339,7 @@ struct GetKeyServerLocationsReply {
 
 	template <class Ar>
 	void serialize(Ar& ar) {
-		serializer(ar, results, resultsTssMapping, tenantEntry, arena, resultsTagMapping);
+		serializer(ar, results, resultsTssMapping, tenantEntry, resultsTagMapping, arena);
 	}
 };
 
@@ -543,7 +543,7 @@ struct ProxySnapRequest {
 
 	template <class Ar>
 	void serialize(Ar& ar) {
-		serializer(ar, snapPayload, snapUID, reply, arena, debugID);
+		serializer(ar, snapPayload, snapUID, reply, debugID, arena);
 	}
 };
 
diff --git a/fdbclient/include/fdbclient/EncryptKeyProxyInterface.h b/fdbclient/include/fdbclient/EncryptKeyProxyInterface.h
index f027b76f7a..3a88a855e0 100644
--- a/fdbclient/include/fdbclient/EncryptKeyProxyInterface.h
+++ b/fdbclient/include/fdbclient/EncryptKeyProxyInterface.h
@@ -132,7 +132,7 @@ struct EKPGetBaseCipherKeysByIdsReply {
 
 	template <class Ar>
 	void serialize(Ar& ar) {
-		serializer(ar, arena, baseCipherDetails, numHits, error);
+		serializer(ar, baseCipherDetails, numHits, error, arena);
 	}
 };
 
diff --git a/fdbclient/include/fdbclient/StorageServerInterface.h b/fdbclient/include/fdbclient/StorageServerInterface.h
index 00e9ff2aef..6b3e64d0aa 100644
--- a/fdbclient/include/fdbclient/StorageServerInterface.h
+++ b/fdbclient/include/fdbclient/StorageServerInterface.h
@@ -416,8 +416,8 @@ struct GetKeyValuesRequest : TimedRequest {
 		           spanContext,
 		           tenantInfo,
 		           options,
-		           arena,
-		           ssLatestCommitVersions);
+		           ssLatestCommitVersions,
+		           arena);
 	}
 };
 
@@ -474,9 +474,9 @@ struct GetMappedKeyValuesRequest : TimedRequest {
 		           spanContext,
 		           tenantInfo,
 		           options,
-		           arena,
 		           ssLatestCommitVersions,
-		           matchIndex);
+		           matchIndex,
+		           arena);
 	}
 };
 
@@ -539,8 +539,8 @@ struct GetKeyValuesStreamRequest {
 		           spanContext,
 		           tenantInfo,
 		           options,
-		           arena,
-		           ssLatestCommitVersions);
+		           ssLatestCommitVersions,
+		           arena);
 	}
 };
 
@@ -588,7 +588,7 @@ struct GetKeyRequest : TimedRequest {
 
 	template <class Ar>
 	void serialize(Ar& ar) {
-		serializer(ar, sel, version, tags, reply, spanContext, tenantInfo, options, arena, ssLatestCommitVersions);
+		serializer(ar, sel, version, tags, reply, spanContext, tenantInfo, options, ssLatestCommitVersions, arena);
 	}
 };
 
@@ -758,7 +758,7 @@ struct SplitMetricsRequest {
 
 	template <class Ar>
 	void serialize(Ar& ar) {
-		serializer(ar, keys, limits, used, estimated, isLastShard, reply, arena, minSplitBytes);
+		serializer(ar, keys, limits, used, estimated, isLastShard, reply, minSplitBytes, arena);
 	}
 };
 
@@ -1038,7 +1038,7 @@ struct OverlappingChangeFeedsReply {
 
 	template <class Ar>
 	void serialize(Ar& ar) {
-		serializer(ar, feeds, arena, feedMetadataVersion);
+		serializer(ar, feeds, feedMetadataVersion, arena);
 	}
 };
 
diff --git a/fdbserver/include/fdbserver/DataDistributorInterface.h b/fdbserver/include/fdbserver/DataDistributorInterface.h
index a7bfcbe1e6..93ed3e358f 100644
--- a/fdbserver/include/fdbserver/DataDistributorInterface.h
+++ b/fdbserver/include/fdbserver/DataDistributorInterface.h
@@ -120,7 +120,7 @@ struct DistributorSnapRequest {
 
 	template <class Ar>
 	void serialize(Ar& ar) {
-		serializer(ar, snapPayload, snapUID, reply, arena, debugID);
+		serializer(ar, snapPayload, snapUID, reply, debugID, arena);
 	}
 };
 
diff --git a/fdbserver/include/fdbserver/ResolverInterface.h b/fdbserver/include/fdbserver/ResolverInterface.h
index 51110e5c01..b746aa8fe6 100644
--- a/fdbserver/include/fdbserver/ResolverInterface.h
+++ b/fdbserver/include/fdbserver/ResolverInterface.h
@@ -139,10 +139,10 @@ struct ResolveTransactionBatchRequest {
 		           transactions,
 		           txnStateTransactions,
 		           reply,
-		           arena,
 		           debugID,
 		           writtenTags,
-		           spanContext);
+		           spanContext,
+		           arena);
 	}
 };
 
diff --git a/fdbserver/include/fdbserver/TLogInterface.h b/fdbserver/include/fdbserver/TLogInterface.h
index 1e101ac16b..cc376fc595 100644
--- a/fdbserver/include/fdbserver/TLogInterface.h
+++ b/fdbserver/include/fdbserver/TLogInterface.h
@@ -328,10 +328,10 @@ struct TLogCommitRequest {
 		           minKnownCommittedVersion,
 		           messages,
 		           reply,
-		           arena,
 		           debugID,
 		           tLogCount,
-		           spanContext);
+		           spanContext,
+		           arena);
 	}
 };
 
diff --git a/flow/flat_buffers.cpp b/flow/flat_buffers.cpp
index ea1c1c2dce..bf7948e024 100644
--- a/flow/flat_buffers.cpp
+++ b/flow/flat_buffers.cpp
@@ -473,13 +473,13 @@ TEST_CASE("/flow/FlatBuffers/VectorRef") {
 				vec.push_back(arena, str);
 			}
 			ObjectWriter writer(Unversioned());
-			writer.serialize(FileIdentifierFor<decltype(vec)>::value, arena, vec);
+			writer.serialize(FileIdentifierFor<decltype(vec)>::value, vec);
 			serializedVector = StringRef(readerArena, writer.toStringRef());
 		}
 		ArenaObjectReader reader(readerArena, serializedVector, Unversioned());
-		// The VectorRef and Arena arguments are intentionally in a different order from the serialize call above.
+		// The Arena argument is intentionally missing from the serialize call above.
 		// Arenas need to get serialized after any Ref types whose memory they own. In order for schema evolution to be
-		// possible, it needs to be okay to reorder an Arena so that it appears after a newly added Ref type. For this
+		// possible, it needs to be okay to move/add an Arena so that it appears after a newly added Ref type. For this
 		// reason, Arenas are ignored by the wire protocol entirely. We test that behavior here.
 		reader.deserialize(FileIdentifierFor<decltype(outVec)>::value, outVec, vecArena);
 	}
diff --git a/flow/include/flow/Arena.h b/flow/include/flow/Arena.h
index 63c50a807b..0c632e8bbd 100644
--- a/flow/include/flow/Arena.h
+++ b/flow/include/flow/Arena.h
@@ -100,6 +100,7 @@ FDB_DECLARE_BOOLEAN_PARAM(FastInaccurateEstimate);
 // memory is freed by deleting the entire Arena at once. See flow/README.md for details on using Arenas.
 class Arena {
 public:
+	constexpr static auto fb_must_appear_last = true;
 	Arena();
 	explicit Arena(size_t reservedSize);
 	//~Arena();
@@ -365,6 +366,8 @@ class Standalone : private Arena, public T {
 public:
 	using RefType = T;
 
+	constexpr static auto fb_must_appear_last = false;
+
 	// T must have no destructor
 	Arena& arena() { return *(Arena*)this; }
 	const Arena& arena() const { return *(const Arena*)this; }
diff --git a/flow/include/flow/ObjectSerializerTraits.h b/flow/include/flow/ObjectSerializerTraits.h
index d396bd5808..964c56e516 100644
--- a/flow/include/flow/ObjectSerializerTraits.h
+++ b/flow/include/flow/ObjectSerializerTraits.h
@@ -37,11 +37,6 @@ struct is_fb_function_t<T, typename std::enable_if<T::is_fb_visitor>::type> : st
 template <class T>
 constexpr bool is_fb_function = is_fb_function_t<T>::value;
 
-template <class Visitor, class... Items>
-typename std::enable_if<is_fb_function<Visitor>, void>::type serializer(Visitor& visitor, Items&... items) {
-	visitor(items...);
-}
-
 template <class... Ts>
 struct pack {};
 
@@ -61,6 +56,40 @@ struct index_impl<0, pack<T, Ts...>> {
 template <int i, class Pack>
 using index_t = typename index_impl<i, Pack>::type;
 
+template <class T, typename = void>
+struct fb_must_appear_last_t : std::false_type {};
+
+template <class T>
+struct fb_must_appear_last_t<T, typename std::enable_if<T::fb_must_appear_last>::type>
+  : std::conditional_t<T::fb_must_appear_last, std::true_type, std::false_type> {};
+
+template <class T>
+constexpr bool fb_must_appear_last = fb_must_appear_last_t<T>::value;
+
+template <class Item, class... Items>
+constexpr bool fb_appears_last_property_helper(pack<Item, Items...>) {
+	if constexpr (sizeof...(Items) == 0) {
+		return true;
+	} else {
+		return !fb_must_appear_last<Item> && fb_appears_last_property_helper(pack<Items...>{});
+	}
+}
+template <class... Items>
+constexpr bool fb_appears_last_property(pack<Items...>) {
+	if constexpr (sizeof...(Items) == 0) {
+		return true;
+	} else {
+		return fb_appears_last_property_helper(pack<Items...>{});
+	}
+}
+
+template <class Visitor, class... Items>
+typename std::enable_if<is_fb_function<Visitor>, void>::type serializer(Visitor& visitor, Items&... items) {
+	static_assert(fb_appears_last_property(pack<Items...>{}),
+	              "An argument to a serializer call that must appear last (Arena?) does not appear last");
+	visitor(items...);
+}
+
 template <class T, typename = void>
 struct scalar_traits : std::false_type {
 	constexpr static size_t size = 0;
diff --git a/flow/include/flow/serialize.h b/flow/include/flow/serialize.h
index 97417c50c3..513340fc38 100644
--- a/flow/include/flow/serialize.h
+++ b/flow/include/flow/serialize.h
@@ -91,6 +91,8 @@ inline typename Archive::READER& operator>>(Archive& ar, Item& item) {
 
 template <class Archive, class Item, class... Items>
 typename Archive::WRITER& serializer(Archive& ar, const Item& item, const Items&... items) {
+	static_assert(fb_appears_last_property(pack<Item, Items...>{}),
+	              "An argument to a serializer call that must appear last (Arena?) does not appear last");
 	save(ar, item);
 	if constexpr (sizeof...(Items) > 0) {
 		serializer(ar, items...);
@@ -100,6 +102,8 @@ typename Archive::WRITER& serializer(Archive& ar, const Item& item, const Items&
 
 template <class Archive, class Item, class... Items>
 typename Archive::READER& serializer(Archive& ar, Item& item, Items&... items) {
+	static_assert(fb_appears_last_property(pack<Item, Items...>{}),
+	              "An argument to a serializer call that must appear last (Arena?) does not appear last");
 	load(ar, item);
 	if constexpr (sizeof...(Items) > 0) {
 		serializer(ar, items...);

From 3887ed5409e1a52ea719fd32269cbe2b9e9cbc6d Mon Sep 17 00:00:00 2001
From: Josh Slocum <josh.slocum@snowflake.com>
Date: Thu, 8 Sep 2022 16:32:24 -0700
Subject: [PATCH 5/6] fixing memory leak caused by uncessary notified version
 promises (#8106)

---
 fdbserver/BlobWorker.actor.cpp | 5 ++++-
 1 file changed, 4 insertions(+), 1 deletion(-)

diff --git a/fdbserver/BlobWorker.actor.cpp b/fdbserver/BlobWorker.actor.cpp
index 5ef6fb92f6..9bb22f9e6a 100644
--- a/fdbserver/BlobWorker.actor.cpp
+++ b/fdbserver/BlobWorker.actor.cpp
@@ -1873,6 +1873,7 @@ ACTOR Future<Void> blobGranuleUpdateFiles(Reference<BlobWorkerData> bwData,
 	state int pendingSnapshots = 0;
 	state Version lastForceFlushVersion = invalidVersion;
 	state std::deque<Version> forceFlushVersions;
+	state Future<Void> nextForceFlush = metadata->forceFlushVersion.whenAtLeast(1);
 
 	state std::deque<std::pair<Version, Version>> rollbacksInProgress;
 	state std::deque<std::pair<Version, Version>> rollbacksCompleted;
@@ -2113,7 +2114,7 @@ ACTOR Future<Void> blobGranuleUpdateFiles(Reference<BlobWorkerData> bwData,
 						}
 					}
 					when(wait(inFlightFiles.empty() ? Never() : success(inFlightFiles.front().future))) {}
-					when(wait(metadata->forceFlushVersion.whenAtLeast(lastForceFlushVersion + 1))) {
+					when(wait(nextForceFlush)) {
 						if (forceFlushVersions.empty() ||
 						    forceFlushVersions.back() < metadata->forceFlushVersion.get()) {
 							forceFlushVersions.push_back(metadata->forceFlushVersion.get());
@@ -2121,6 +2122,7 @@ ACTOR Future<Void> blobGranuleUpdateFiles(Reference<BlobWorkerData> bwData,
 						if (metadata->forceFlushVersion.get() > lastForceFlushVersion) {
 							lastForceFlushVersion = metadata->forceFlushVersion.get();
 						}
+						nextForceFlush = metadata->forceFlushVersion.whenAtLeast(lastForceFlushVersion + 1);
 					}
 				}
 			} catch (Error& e) {
@@ -2255,6 +2257,7 @@ ACTOR Future<Void> blobGranuleUpdateFiles(Reference<BlobWorkerData> bwData,
 									forceFlushVersions.clear();
 									lastForceFlushVersion = 0;
 									metadata->forceFlushVersion = NotifiedVersion();
+									nextForceFlush = metadata->forceFlushVersion.whenAtLeast(1);
 
 									Reference<ChangeFeedData> cfData =
 									    makeReference<ChangeFeedData>(bwData->db.getPtr());

From e66015cbe4408d2db6bcafbe95ffb2ca7d9b4396 Mon Sep 17 00:00:00 2001
From: Josh Slocum <josh.slocum@snowflake.com>
Date: Thu, 8 Sep 2022 16:37:44 -0700
Subject: [PATCH 6/6] Including change feed bytes towards storage queue (#8107)

---
 fdbclient/ServerKnobs.cpp                 |  1 +
 fdbclient/include/fdbclient/ServerKnobs.h |  1 +
 fdbserver/storageserver.actor.cpp         | 31 +++++++++++++++++++++++
 3 files changed, 33 insertions(+)

diff --git a/fdbclient/ServerKnobs.cpp b/fdbclient/ServerKnobs.cpp
index da373616d6..0462721816 100644
--- a/fdbclient/ServerKnobs.cpp
+++ b/fdbclient/ServerKnobs.cpp
@@ -642,6 +642,7 @@ void ServerKnobs::initialize(Randomize randomize, ClientKnobs* clientKnobs, IsSi
 	init( STORAGE_HARD_LIMIT_VERSION_OVERAGE, VERSIONS_PER_SECOND / 4.0 );
 	init( STORAGE_DURABILITY_LAG_HARD_MAX,                    2000e6 ); if( smallStorageTarget ) STORAGE_DURABILITY_LAG_HARD_MAX = 100e6;
 	init( STORAGE_DURABILITY_LAG_SOFT_MAX,                     250e6 ); if( smallStorageTarget ) STORAGE_DURABILITY_LAG_SOFT_MAX = 10e6;
+	init( STORAGE_INCLUDE_FEED_STORAGE_QUEUE,                   true ); if ( randomize && BUGGIFY ) STORAGE_INCLUDE_FEED_STORAGE_QUEUE = false;
 
 	//FIXME: Low priority reads are disabled by assigning very high knob values, reduce knobs for 7.0
 	init( LOW_PRIORITY_STORAGE_QUEUE_BYTES,                    775e8 ); if( smallStorageTarget ) LOW_PRIORITY_STORAGE_QUEUE_BYTES = 1750e3;
diff --git a/fdbclient/include/fdbclient/ServerKnobs.h b/fdbclient/include/fdbclient/ServerKnobs.h
index b142ef63c7..d8cb369ab8 100644
--- a/fdbclient/include/fdbclient/ServerKnobs.h
+++ b/fdbclient/include/fdbclient/ServerKnobs.h
@@ -573,6 +573,7 @@ public:
 	int64_t STORAGE_HARD_LIMIT_VERSION_OVERAGE;
 	int64_t STORAGE_DURABILITY_LAG_HARD_MAX;
 	int64_t STORAGE_DURABILITY_LAG_SOFT_MAX;
+	bool STORAGE_INCLUDE_FEED_STORAGE_QUEUE;
 
 	int64_t LOW_PRIORITY_STORAGE_QUEUE_BYTES;
 	int64_t LOW_PRIORITY_DURABILITY_LAG;
diff --git a/fdbserver/storageserver.actor.cpp b/fdbserver/storageserver.actor.cpp
index 4339bf01d9..011f3b714e 100644
--- a/fdbserver/storageserver.actor.cpp
+++ b/fdbserver/storageserver.actor.cpp
@@ -940,6 +940,8 @@ public:
 	std::unordered_map<NetworkAddress, std::map<UID, Version>> changeFeedClientVersions;
 	std::unordered_map<Key, Version> changeFeedCleanupDurable;
 	int64_t activeFeedQueries = 0;
+	int64_t changeFeedMemoryBytes = 0;
+	std::deque<std::pair<Version, int64_t>> feedMemoryBytesByVersion;
 
 	// newestAvailableVersion[k]
 	//   == invalidVersion -> k is unavailable at all versions
@@ -1201,6 +1203,7 @@ public:
 			specialCounter(cc, "KvstoreInlineKey", [self]() { return std::get<2>(self->storage.getSize()); });
 			specialCounter(cc, "ActiveChangeFeeds", [self]() { return self->uidChangeFeed.size(); });
 			specialCounter(cc, "ActiveChangeFeedQueries", [self]() { return self->activeFeedQueries; });
+			specialCounter(cc, "ChangeFeedMemoryBytes", [self]() { return self->changeFeedMemoryBytes; });
 		}
 	} counters;
 
@@ -1440,6 +1443,20 @@ public:
 		return minVersion;
 	}
 
+	// count in-memory change feed bytes towards storage queue size, for the purposes of memory management and
+	// throttling
+	void addFeedBytesAtVersion(int64_t bytes, Version version) {
+		if (feedMemoryBytesByVersion.empty() || version != feedMemoryBytesByVersion.back().first) {
+			ASSERT(feedMemoryBytesByVersion.empty() || version >= feedMemoryBytesByVersion.back().first);
+			feedMemoryBytesByVersion.push_back({ version, 0 });
+		}
+		feedMemoryBytesByVersion.back().second += bytes;
+		changeFeedMemoryBytes += bytes;
+		if (SERVER_KNOBS->STORAGE_INCLUDE_FEED_STORAGE_QUEUE) {
+			counters.bytesInput += bytes;
+		}
+	}
+
 	void getSplitPoints(SplitRangeRequest const& req) {
 		try {
 			Optional<TenantMapEntry> entry = getTenantEntry(version.get(), req.tenantInfo);
@@ -5076,6 +5093,17 @@ bool changeDurableVersion(StorageServer* data, Version desiredDurableVersion) {
 		data->counters.bytesDurable += bytesDurable;
 	}
 
+	int64_t feedBytesDurable = 0;
+	while (!data->feedMemoryBytesByVersion.empty() &&
+	       data->feedMemoryBytesByVersion.front().first <= desiredDurableVersion) {
+		feedBytesDurable += data->feedMemoryBytesByVersion.front().second;
+		data->feedMemoryBytesByVersion.pop_front();
+	}
+	data->changeFeedMemoryBytes -= feedBytesDurable;
+	if (SERVER_KNOBS->STORAGE_INCLUDE_FEED_STORAGE_QUEUE) {
+		data->counters.bytesDurable += feedBytesDurable;
+	}
+
 	if (EXPENSIVE_VALIDATION) {
 		// Check that the above loop did its job
 		auto view = data->data().atLatest();
@@ -5261,6 +5289,7 @@ void applyChangeFeedMutation(StorageServer* self, MutationRef const& m, Version
 				}
 				it->mutations.back().mutations.push_back_deep(it->mutations.back().arena(), m);
 				self->currentChangeFeeds.insert(it->id);
+				self->addFeedBytesAtVersion(m.totalSize(), version);
 
 				DEBUG_MUTATION("ChangeFeedWriteSet", version, m, self->thisServerID)
 				    .detail("Range", it->range)
@@ -5289,6 +5318,8 @@ void applyChangeFeedMutation(StorageServer* self, MutationRef const& m, Version
 					}
 					it->mutations.back().mutations.push_back_deep(it->mutations.back().arena(), m);
 					self->currentChangeFeeds.insert(it->id);
+					self->addFeedBytesAtVersion(m.totalSize(), version);
+
 					DEBUG_MUTATION("ChangeFeedWriteClear", version, m, self->thisServerID)
 					    .detail("Range", it->range)
 					    .detail("ChangeFeedID", it->id);