[Databuckets] Improved Reliability and Performance of Databuckets (#4562)

* [Databuckets] Don't broadcast client-scoped updates

* Remove temp feature flag

* Remove distributed caching, only cache for character scoped data, simplify

* Update bot.cpp

* Cleanup

* Update data_bucket.cpp

* Cleanup

* Cleanup

* Remove BulkLoadEntities from LoadNPCTypes

* Update data_bucket.cpp

* Cleanup

* More cleanup

* More cleanup

* BulkLoadEntities to BulkLoadEntitiesToCache

* Add CanCache in DeleteData to gate an unnecessary call

zone/data_bucket.cpp

@@ -8,7 +8,7 @@
 
 extern WorldServer worldserver;
 
-std::vector<DataBucketCacheEntry> g_data_bucket_cache = {};
+std::vector<DataBucketsRepository::DataBuckets> g_data_bucket_cache = {};
 
 void DataBucket::SetData(const std::string &bucket_key, const std::string &bucket_value, std::string expires_time)
 {
@@ -58,14 +58,14 @@ void DataBucket::SetData(const DataBucketKey &k)
 	b.value   = k.value;
 
 	if (bucket_id) {
-		// loop cache and update cache value and timestamp
-		for (auto &ce: g_data_bucket_cache) {
-			if (CheckBucketMatch(ce.e, k)) {
-				ce.e             = b;
-				ce.updated_time  = GetCurrentTimeUNIX();
-				ce.update_action = DataBucketCacheUpdateAction::Upsert;
-				SendDataBucketCacheUpdate(ce);
-				break;
+
+		// update the cache if it exists
+		if (CanCache(k)) {
+			for (auto &e: g_data_bucket_cache) {
+				if (CheckBucketMatch(e, k)) {
+					e = b;
+					break;
+				}
 			}
 		}
 
@@ -74,28 +74,18 @@ void DataBucket::SetData(const DataBucketKey &k)
 	else {
 		b.key_ = k.key;
 		b = DataBucketsRepository::InsertOne(database, b);
-		if (!ExistsInCache(b)) {
-			// add data bucket and timestamp to cache
-			auto ce = DataBucketCacheEntry{
-				.e = b,
-				.updated_time = DataBucket::GetCurrentTimeUNIX(),
-				.update_action = DataBucketCacheUpdateAction::Upsert
-			};
-
-			g_data_bucket_cache.emplace_back(ce);
-
-			SendDataBucketCacheUpdate(ce);
 
+		// add to cache if it doesn't exist
+		if (CanCache(k) && !ExistsInCache(b)) {
 			DeleteFromMissesCache(b);
+			g_data_bucket_cache.emplace_back(b);
 		}
 	}
 }
 
 std::string DataBucket::GetData(const std::string &bucket_key)
 {
-	DataBucketKey k = {};
-	k.key = bucket_key;
-	return GetData(k).value;
+	return GetData(DataBucketKey{.key = bucket_key}).value;
 }
 
 // GetData fetches bucket data from the database or cache if it exists
@@ -112,22 +102,27 @@ DataBucketsRepository::DataBuckets DataBucket::GetData(const DataBucketKey &k, b
 		k.npc_id
 	);
 
-	for (const auto &ce: g_data_bucket_cache) {
-		if (CheckBucketMatch(ce.e, k)) {
-			if (ce.e.expires > 0 && ce.e.expires < std::time(nullptr)) {
-				LogDataBuckets("Attempted to read expired key [{}] removing from cache", ce.e.key_);
-				DeleteData(k);
-				return DataBucketsRepository::NewEntity();
-			}
+	bool can_cache = CanCache(k);
 
-			// this is a bucket miss, return empty entity
-			// we still cache bucket misses, so we don't have to hit the database
-			if (ce.e.id == 0) {
-				return DataBucketsRepository::NewEntity();
-			}
+	// check the cache first if we can cache
+	if (can_cache) {
+		for (const auto &e: g_data_bucket_cache) {
+			if (CheckBucketMatch(e, k)) {
+				if (e.expires > 0 && e.expires < std::time(nullptr)) {
+					LogDataBuckets("Attempted to read expired key [{}] removing from cache", e.key_);
+					DeleteData(k);
+					return DataBucketsRepository::NewEntity();
+				}
 
-			LogDataBuckets("Returning key [{}] value [{}] from cache", ce.e.key_, ce.e.value);
-			return ce.e;
+				// this is a bucket miss, return empty entity
+				// we still cache bucket misses, so we don't have to hit the database
+				if (e.id == 0) {
+					return DataBucketsRepository::NewEntity();
+				}
+
+				LogDataBuckets("Returning key [{}] value [{}] from cache", e.key_, e.value);
+				return e;
+			}
 		}
 	}
 
@@ -144,23 +139,21 @@ DataBucketsRepository::DataBuckets DataBucket::GetData(const DataBucketKey &k, b
 
 		// if we're ignoring the misses cache, don't add to the cache
 		// the only place this is ignored is during the initial read of SetData
-		if (!ignore_misses_cache) {
+		bool add_to_misses_cache = !ignore_misses_cache && can_cache;
+		if (add_to_misses_cache) {
 			size_t size_before = g_data_bucket_cache.size();
 
 			// cache bucket misses, so we don't have to hit the database
 			// when scripts try to read a bucket that doesn't exist
 			g_data_bucket_cache.emplace_back(
-				DataBucketCacheEntry{
-					.e = DataBucketsRepository::DataBuckets{
-						.id = 0,
-						.key_ = k.key,
-						.value = "",
-						.expires = 0,
-						.character_id = k.character_id,
-						.npc_id = k.npc_id,
-						.bot_id = k.bot_id
-					},
-					.updated_time = DataBucket::GetCurrentTimeUNIX()
+				DataBucketsRepository::DataBuckets{
+					.id = 0,
+					.key_ = k.key,
+					.value = "",
+					.expires = 0,
+					.character_id = k.character_id,
+					.npc_id = k.npc_id,
+					.bot_id = k.bot_id
 				}
 			);
 
@@ -178,60 +171,54 @@ DataBucketsRepository::DataBuckets DataBucket::GetData(const DataBucketKey &k, b
 		return {};
 	}
 
+	auto bucket = r.front();
+
 	// if the entry has expired, delete it
-	if (r[0].expires > 0 && r[0].expires < (long long) std::time(nullptr)) {
+	if (bucket.expires > 0 && bucket.expires < (long long) std::time(nullptr)) {
 		DeleteData(k);
 		return {};
 	}
 
-	bool      has_cache = false;
-	for (auto &ce: g_data_bucket_cache) {
-		if (ce.e.id == r[0].id) {
-			has_cache = true;
-			break;
+	// add to cache if it doesn't exist
+	if (can_cache) {
+		bool has_cache = false;
+
+		for (auto &e: g_data_bucket_cache) {
+			if (e.id == bucket.id) {
+				has_cache = true;
+				break;
+			}
+		}
+
+		if (!has_cache) {
+			g_data_bucket_cache.emplace_back(bucket);
 		}
 	}
 
-	if (!has_cache) {
-		// add data bucket and timestamp to cache
-		g_data_bucket_cache.emplace_back(
-			DataBucketCacheEntry{
-				.e = r[0],
-				.updated_time = DataBucket::GetCurrentTimeUNIX()
-			}
-		);
-	}
-
-	return r[0];
+	return bucket;
 }
 
 std::string DataBucket::GetDataExpires(const std::string &bucket_key)
 {
-	DataBucketKey k = {};
-	k.key = bucket_key;
-
-	return GetDataExpires(k);
+	return GetDataExpires(DataBucketKey{.key = bucket_key});
 }
 
 std::string DataBucket::GetDataRemaining(const std::string &bucket_key)
 {
-	DataBucketKey k = {};
-	k.key = bucket_key;
-	return GetDataRemaining(k);
+	return GetDataRemaining(DataBucketKey{.key = bucket_key});
 }
 
 bool DataBucket::DeleteData(const std::string &bucket_key)
 {
-	DataBucketKey k = {};
-	k.key = bucket_key;
-	return DeleteData(k);
+	return DeleteData(DataBucketKey{.key = bucket_key});
 }
 
 // GetDataBuckets bulk loads all data buckets for a mob
 bool DataBucket::GetDataBuckets(Mob *mob)
 {
-	DataBucketLoadType::Type t;
-	const uint32             id = mob->GetMobTypeIdentifier();
+	DataBucketLoadType::Type t{};
+
+	const uint32 id = mob->GetMobTypeIdentifier();
 
 	if (!id) {
 		return false;
@@ -243,46 +230,39 @@ bool DataBucket::GetDataBuckets(Mob *mob)
 	else if (mob->IsClient()) {
 		t = DataBucketLoadType::Client;
 	}
-	else if (mob->IsNPC()) {
-		t = DataBucketLoadType::NPC;
-	}
 
-	BulkLoadEntities(t, {id});
+	BulkLoadEntitiesToCache(t, {id});
 
 	return true;
 }
 
 bool DataBucket::DeleteData(const DataBucketKey &k)
 {
-	size_t size_before = g_data_bucket_cache.size();
+	if (CanCache(k)) {
+		size_t size_before = g_data_bucket_cache.size();
 
-	// delete from cache where contents match
-	g_data_bucket_cache.erase(
-		std::remove_if(
-			g_data_bucket_cache.begin(),
-			g_data_bucket_cache.end(),
-			[&](DataBucketCacheEntry &ce) {
-				bool match = CheckBucketMatch(ce.e, k);
-				if (match) {
-					ce.update_action = DataBucketCacheUpdateAction::Delete;
-					SendDataBucketCacheUpdate(ce);
+		// delete from cache where contents match
+		g_data_bucket_cache.erase(
+			std::remove_if(
+				g_data_bucket_cache.begin(),
+				g_data_bucket_cache.end(),
+				[&](DataBucketsRepository::DataBuckets &e) {
+					return CheckBucketMatch(e, k);
 				}
+			),
+			g_data_bucket_cache.end()
+		);
 
-				return match;
-			}
-		),
-		g_data_bucket_cache.end()
-	);
-
-	LogDataBuckets(
-		"Deleting bucket key [{}] bot_id [{}] character_id [{}] npc_id [{}] cache size before [{}] after [{}]",
-		k.key,
-		k.bot_id,
-		k.character_id,
-		k.npc_id,
-		size_before,
-		g_data_bucket_cache.size()
-	);
+		LogDataBuckets(
+			"Deleting bucket key [{}] bot_id [{}] character_id [{}] npc_id [{}] cache size before [{}] after [{}]",
+			k.key,
+			k.bot_id,
+			k.character_id,
+			k.npc_id,
+			size_before,
+			g_data_bucket_cache.size()
+		);
+	}
 
 	return DataBucketsRepository::DeleteWhere(
 		database,
@@ -371,23 +351,21 @@ bool DataBucket::CheckBucketMatch(const DataBucketsRepository::DataBuckets &dbe,
 	);
 }
 
-void DataBucket::BulkLoadEntities(DataBucketLoadType::Type t, std::vector<uint32> ids)
+void DataBucket::BulkLoadEntitiesToCache(DataBucketLoadType::Type t, std::vector<uint32> ids)
 {
 	if (ids.empty()) {
 		return;
 	}
 
 	if (ids.size() == 1) {
-		bool            has_cache = false;
-		for (const auto &ce: g_data_bucket_cache) {
+		bool has_cache = false;
+
+		for (const auto &e: g_data_bucket_cache) {
 			if (t == DataBucketLoadType::Bot) {
-				has_cache = ce.e.bot_id == ids[0];
+				has_cache = e.bot_id == ids[0];
 			}
 			else if (t == DataBucketLoadType::Client) {
-				has_cache = ce.e.character_id == ids[0];
-			}
-			else if (t == DataBucketLoadType::NPC) {
-				has_cache = ce.e.npc_id == ids[0];
+				has_cache = e.character_id == ids[0];
 			}
 		}
 
@@ -406,9 +384,6 @@ void DataBucket::BulkLoadEntities(DataBucketLoadType::Type t, std::vector<uint32
 		case DataBucketLoadType::Client:
 			column = "character_id";
 			break;
-		case DataBucketLoadType::NPC:
-			column = "npc_id";
-			break;
 		default:
 			LogError("Incorrect LoadType [{}]", static_cast<int>(t));
 			break;
@@ -442,12 +417,7 @@ void DataBucket::BulkLoadEntities(DataBucketLoadType::Type t, std::vector<uint32
 		if (!ExistsInCache(e)) {
 			LogDataBucketsDetail("bucket id [{}] bucket key [{}] bucket value [{}]", e.id, e.key_, e.value);
 
-			g_data_bucket_cache.emplace_back(
-				DataBucketCacheEntry{
-					.e = e,
-					.updated_time = GetCurrentTimeUNIX()
-				}
-			);
+			g_data_bucket_cache.emplace_back(e);
 		}
 	}
 
@@ -461,7 +431,7 @@ void DataBucket::BulkLoadEntities(DataBucketLoadType::Type t, std::vector<uint32
 	);
 }
 
-void DataBucket::DeleteCachedBuckets(DataBucketLoadType::Type t, uint32 id)
+void DataBucket::DeleteCachedBuckets(DataBucketLoadType::Type type, uint32 id)
 {
 	size_t size_before = g_data_bucket_cache.size();
 
@@ -469,11 +439,10 @@ void DataBucket::DeleteCachedBuckets(DataBucketLoadType::Type t, uint32 id)
 		std::remove_if(
 			g_data_bucket_cache.begin(),
 			g_data_bucket_cache.end(),
-			[&](DataBucketCacheEntry &ce) {
+			[&](DataBucketsRepository::DataBuckets &e) {
 				return (
-					(t == DataBucketLoadType::Bot && ce.e.bot_id == id) ||
-					(t == DataBucketLoadType::Client && ce.e.character_id == id) ||
-					(t == DataBucketLoadType::NPC && ce.e.npc_id == id)
+					(type == DataBucketLoadType::Bot && e.bot_id == id) ||
+					(type == DataBucketLoadType::Client && e.character_id == id)
 				);
 			}
 		),
@@ -482,24 +451,17 @@ void DataBucket::DeleteCachedBuckets(DataBucketLoadType::Type t, uint32 id)
 
 	LogDataBuckets(
 		"LoadType [{}] id [{}] cache size before [{}] after [{}]",
-		DataBucketLoadType::Name[t],
+		DataBucketLoadType::Name[type],
 		id,
 		size_before,
 		g_data_bucket_cache.size()
 	);
 }
 
-int64_t DataBucket::GetCurrentTimeUNIX()
+bool DataBucket::ExistsInCache(const DataBucketsRepository::DataBuckets &entry)
 {
-	return std::chrono::duration_cast<std::chrono::nanoseconds>(
-		std::chrono::system_clock::now().time_since_epoch()
-	).count();
-}
-
-bool DataBucket::ExistsInCache(const DataBucketsRepository::DataBuckets &e)
-{
-	for (const auto &ce: g_data_bucket_cache) {
-		if (ce.e.id == e.id) {
+	for (const auto &e: g_data_bucket_cache) {
+		if (e.id == entry.id) {
 			return true;
 		}
 	}
@@ -507,134 +469,6 @@ bool DataBucket::ExistsInCache(const DataBucketsRepository::DataBuckets &e)
 	return false;
 }
 
-bool DataBucket::SendDataBucketCacheUpdate(const DataBucketCacheEntry &e)
-{
-	if (!e.e.id) {
-		return false;
-	}
-
-	EQ::Net::DynamicPacket p;
-	p.PutSerialize(0, e);
-
-	auto pack_size = sizeof(ServerDataBucketCacheUpdate_Struct) + p.Length();
-	auto pack      = new ServerPacket(ServerOP_DataBucketCacheUpdate, static_cast<uint32_t>(pack_size));
-	auto buf       = reinterpret_cast<ServerDataBucketCacheUpdate_Struct *>(pack->pBuffer);
-
-	buf->cereal_size = static_cast<uint32_t>(p.Length());
-
-	memcpy(buf->cereal_data, p.Data(), p.Length());
-
-	worldserver.SendPacket(pack);
-
-	return true;
-}
-
-void DataBucket::HandleWorldMessage(ServerPacket *p)
-{
-	DataBucketCacheEntry         n;
-	auto                         s = (ServerDataBucketCacheUpdate_Struct *) p->pBuffer;
-	EQ::Util::MemoryStreamReader ss(s->cereal_data, s->cereal_size);
-	cereal::BinaryInputArchive   archive(ss);
-	archive(n);
-
-	LogDataBucketsDetail(
-		"Received cache packet for id [{}] key [{}] value [{}] action [{}]",
-		n.e.id,
-		n.e.key_,
-		n.e.value,
-		static_cast<int>(n.update_action)
-	);
-
-	// delete
-	if (n.update_action == DataBucketCacheUpdateAction::Delete) {
-		DeleteFromMissesCache(n.e);
-
-		g_data_bucket_cache.erase(
-			std::remove_if(
-				g_data_bucket_cache.begin(),
-				g_data_bucket_cache.end(),
-				[&](DataBucketCacheEntry &ce) {
-					bool match = n.e.id > 0 && ce.e.id == n.e.id;
-					if (match) {
-						LogDataBuckets(
-							"[delete] cache key [{}] id [{}] cache_size before [{}] after [{}]",
-							ce.e.key_,
-							ce.e.id,
-							g_data_bucket_cache.size(),
-							g_data_bucket_cache.size() - 1
-						);
-					}
-					return match;
-				}
-			),
-			g_data_bucket_cache.end()
-		);
-
-		return;
-	}
-
-	// update
-	bool has_key = false;
-	for (auto &ce: g_data_bucket_cache) {
-		// update cache
-		if (ce.e.id == n.e.id) {
-			// reject old updates
-			int64 time_delta = ce.updated_time - n.updated_time;
-			if (ce.updated_time >= n.updated_time) {
-				LogDataBuckets(
-					"Attempted to update older cache key [{}] rejecting old time [{}] new time [{}] delta [{}] cache_size [{}]",
-					ce.e.key_,
-					ce.updated_time,
-					n.updated_time,
-					time_delta,
-					g_data_bucket_cache.size()
-				);
-				return;
-			}
-
-			DeleteFromMissesCache(n.e);
-
-			LogDataBuckets(
-				"[update] cache id [{}] key [{}] value [{}] old time [{}] new time [{}] delta [{}] cache_size [{}]",
-				ce.e.id,
-				ce.e.key_,
-				n.e.value,
-				ce.updated_time,
-				n.updated_time,
-				time_delta,
-				g_data_bucket_cache.size()
-			);
-			ce.e            = n.e;
-			ce.updated_time = n.updated_time;
-			has_key = true;
-			break;
-		}
-	}
-
-	// create
-	if (!has_key) {
-		DeleteFromMissesCache(n.e);
-
-		size_t size_before = g_data_bucket_cache.size();
-
-		g_data_bucket_cache.emplace_back(
-			DataBucketCacheEntry{
-				.e = n.e,
-				.updated_time = GetCurrentTimeUNIX()
-			}
-		);
-
-		LogDataBuckets(
-			"[create] Adding new cache id [{}] key [{}] value [{}] cache size before [{}] after [{}]",
-			n.e.id,
-			n.e.key_,
-			n.e.value,
-			size_before,
-			g_data_bucket_cache.size()
-		);
-	}
-}
-
 void DataBucket::DeleteFromMissesCache(DataBucketsRepository::DataBuckets e)
 {
 	// delete from cache where there might have been a written bucket miss to the cache
@@ -645,11 +479,11 @@ void DataBucket::DeleteFromMissesCache(DataBucketsRepository::DataBuckets e)
 		std::remove_if(
 			g_data_bucket_cache.begin(),
 			g_data_bucket_cache.end(),
-			[&](DataBucketCacheEntry &ce) {
-				return ce.e.id == 0 && ce.e.key_ == e.key_ &&
-					   ce.e.character_id == e.character_id &&
-					   ce.e.npc_id == e.npc_id &&
-					   ce.e.bot_id == e.bot_id;
+			[&](DataBucketsRepository::DataBuckets &ce) {
+				return ce.id == 0 && ce.key_ == e.key_ &&
+					   ce.character_id == e.character_id &&
+					   ce.npc_id == e.npc_id &&
+					   ce.bot_id == e.bot_id;
 			}
 		),
 		g_data_bucket_cache.end()
@@ -667,3 +501,47 @@ void DataBucket::ClearCache()
 	g_data_bucket_cache.clear();
 	LogInfo("Cleared data buckets cache");
 }
+
+void DataBucket::DeleteFromCache(uint64 id, DataBucketLoadType::Type type)
+{
+	size_t size_before = g_data_bucket_cache.size();
+
+	g_data_bucket_cache.erase(
+		std::remove_if(
+			g_data_bucket_cache.begin(),
+			g_data_bucket_cache.end(),
+			[&](DataBucketsRepository::DataBuckets &e) {
+				switch (type) {
+					case DataBucketLoadType::Bot:
+						return e.bot_id == id;
+					case DataBucketLoadType::Client:
+						return e.character_id == id;
+					default:
+						return false;
+				}
+			}
+		),
+		g_data_bucket_cache.end()
+	);
+
+	LogDataBuckets(
+		"Deleted [{}] id [{}] from cache size before [{}] after [{}]",
+		DataBucketLoadType::Name[type],
+		id,
+		size_before,
+		g_data_bucket_cache.size()
+	);
+}
+
+// CanCache returns whether a bucket can be cached or not
+// characters are only in one zone at a time so we can cache locally to the zone
+// bots (not implemented) are only in one zone at a time so we can cache locally to the zone
+// npcs (ids) can be in multiple zones so we can't cache locally to the zone
+bool DataBucket::CanCache(const DataBucketKey &key)
+{
+	if (key.character_id > 0 || key.bot_id > 0) {
+		return true;
+	}
+
+	return false;
+}