[Data Buckets] Distributed Databucket Caching (#3500)

* [Data Buckets] Zone-Based Data Bucket Caching # Notes - Adds a data bucket cache so we're not needlessly hitting the database every time we need to read a data bucket value. * Cleanup and unify GetData access patterns * Cache work * Push * Add to cache when we fetch and do a db hit * Handle bucket misses in cache * Formatting * Logging * [Data Buckets] Zone-Based Data Bucket Caching - Adds a data bucket cache so we're not needlessly hitting the database every time we need to read a data bucket value. * Cleanup and unify GetData access patterns * Cache work * Push * Add to cache when we fetch and do a db hit * Handle bucket misses in cache * Formatting * Remove redundant fetches from cache since GetData does the same thing * Push progress * Distributed cache work * Logging * Fix issue with scoping where same named keys could return overlapping results * Misses cache tweak, logging, comments * Add bot, client, and NPC bucket methods to Lua. --------- Co-authored-by: Akkadius <akkadius1@gmail.com>
2026-06-11 15:58:36 +00:00 · 2023-07-24 13:22:50 -04:00
parent 6c2886a71d
commit a75648f73f
23 changed files with 788 additions and 79 deletions
@@ -1,9 +1,14 @@
 #include "data_bucket.h"
+#include "entity.h"
 #include "zonedb.h"
 #include "mob.h"
+#include "worldserver.h"
 #include <ctime>
 #include <cctype>
-#include "../common/repositories/data_buckets_repository.h"
+
+extern WorldServer worldserver;
+
+std::vector<DataBucketCacheEntry> g_data_bucket_cache = {};

 void DataBucket::SetData(const std::string &bucket_key, const std::string &bucket_value, std::string expires_time)
 {
@@ -22,40 +27,67 @@ void DataBucket::SetData(const std::string &bucket_key, const std::string &bucke
 void DataBucket::SetData(const DataBucketKey &k)
 {
 	auto b = DataBucketsRepository::NewEntity();
-	auto r = GetData(k);
+	auto r = GetData(k, true);
 	// if we have an entry, use it
 	if (r.id > 0) {
 		b = r;
 	}

+	// add scoping to bucket
 	if (k.character_id > 0) {
 		b.character_id = k.character_id;
-	} else if (k.npc_id > 0) {
+	}
+	else if (k.npc_id > 0) {
 		b.npc_id = k.npc_id;
-	} else if (k.bot_id > 0) {
+	}
+	else if (k.bot_id > 0) {
 		b.bot_id = k.bot_id;
 	}

-	uint64    bucket_id         = b.id;
-	long long expires_time_unix = 0;
+	const uint64 bucket_id         = b.id;
+	int64        expires_time_unix = 0;

 	if (!k.expires.empty()) {
-		expires_time_unix = (long long) std::time(nullptr) + Strings::ToInt(k.expires);
+		expires_time_unix = static_cast<int64>(std::time(nullptr)) + Strings::ToInt(k.expires);
 		if (isalpha(k.expires[0]) || isalpha(k.expires[k.expires.length() - 1])) {
-			expires_time_unix = (long long) std::time(nullptr) + Strings::TimeToSeconds(k.expires);
+			expires_time_unix = static_cast<int64>(std::time(nullptr)) + Strings::TimeToSeconds(k.expires);
 		}
 	}

-	if (bucket_id > 0) {
-		b.expires = expires_time_unix;
-		b.value   = k.value;
+	b.expires = expires_time_unix;
+	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;
+			}
+		}
+
 		DataBucketsRepository::UpdateOne(database, b);
 	}
 	else {
-		b.expires = expires_time_unix;
-		b.key_    = k.key;
-		b.value   = k.value;
-		DataBucketsRepository::InsertOne(database, b);
+		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);
+
+			DeleteFromMissesCache(b);
+		}
 	}
 }

@@ -66,8 +98,39 @@ std::string DataBucket::GetData(const std::string &bucket_key)
 	return GetData(k).value;
 }

-DataBucketsRepository::DataBuckets DataBucket::GetData(const DataBucketKey &k)
+// GetData fetches bucket data from the database or cache if it exists
+// if the bucket doesn't exist, it will be added to the cache as a miss
+// if ignore_misses_cache is true, the bucket will not be added to the cache as a miss
+// the only place we should be ignoring the misses cache is on the initial read during SetData
+DataBucketsRepository::DataBuckets DataBucket::GetData(const DataBucketKey &k, bool ignore_misses_cache)
 {
+	LogDataBuckets(
+		"Getting bucket key [{}] bot_id [{}] character_id [{}] npc_id [{}]",
+		k.key,
+		k.bot_id,
+		k.character_id,
+		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();
+			}
+
+			// 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();
+			}
+
+			LogDataBuckets("Returning key [{}] value [{}] from cache", ce.e.key_, ce.e.value);
+			return ce.e;
+		}
+	}
+
 	auto r = DataBucketsRepository::GetWhere(
 		database,
 		fmt::format(
@@ -78,6 +141,40 @@ DataBucketsRepository::DataBuckets DataBucket::GetData(const DataBucketKey &k)
 	);

 	if (r.empty()) {
+
+		// 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) {
+			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()
+				}
+			);
+
+			LogDataBuckets(
+				"Key [{}] not found in database, adding to cache as a miss character_id [{}] npc_id [{}] bot_id [{}] cache size before [{}] after [{}]",
+				k.key,
+				k.character_id,
+				k.npc_id,
+				k.bot_id,
+				size_before,
+				g_data_bucket_cache.size()
+			);
+		}
+
 		return {};
 	}

@@ -87,6 +184,24 @@ DataBucketsRepository::DataBuckets DataBucket::GetData(const DataBucketKey &k)
 		return {};
 	}

+	bool      has_cache = false;
+	for (auto &ce: g_data_bucket_cache) {
+		if (ce.e.id == r[0].id) {
+			has_cache = true;
+			break;
+		}
+	}
+
+	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];
 }

@@ -107,57 +222,68 @@ std::string DataBucket::GetDataRemaining(const std::string &bucket_key)

 bool DataBucket::DeleteData(const std::string &bucket_key)
 {
-	DataBucketKey r = {};
-	r.key = bucket_key;
-	return DeleteData(r);
+	DataBucketKey k = {};
+	k.key = bucket_key;
+	return DeleteData(k);
 }

+// GetDataBuckets bulk loads all data buckets for a mob
 bool DataBucket::GetDataBuckets(Mob *mob)
 {
-	DataBucketKey k = mob->GetScopedBucketKeys();
-	auto          l = BaseDataBucketsRepository::GetWhere(
-		database,
-		fmt::format(
-			"{} (`expires` > {} OR `expires` = 0)",
-			DataBucket::GetScopedDbFilters(k),
-			(long long) std::time(nullptr)
-		)
-	);
+	DataBucketLoadType::Type t;
+	const uint32             id = mob->GetMobTypeIdentifier();

-	if (l.empty()) {
+	if (!id) {
 		return false;
 	}

-	mob->m_data_bucket_cache.clear();
-
-	DataBucketCache d;
-
-	for (const auto &e: l) {
-		d.bucket_id      = e.id;
-		d.bucket_key     = e.key_;
-		d.bucket_value   = e.value;
-		d.bucket_expires = e.expires;
-
-		mob->m_data_bucket_cache.emplace_back(d);
+	if (mob->IsBot()) {
+		t = DataBucketLoadType::Bot;
 	}
+	else if (mob->IsClient()) {
+		t = DataBucketLoadType::Client;
+	}
+	else if (mob->IsNPC()) {
+		t = DataBucketLoadType::NPC;
+	}
+
+	BulkLoadEntities(t, {id});

 	return true;
 }

-std::string DataBucket::CheckBucketKey(const Mob *mob, const DataBucketKey &k)
-{
-	std::string     bucket_value;
-	for (const auto &d: mob->m_data_bucket_cache) {
-		if (d.bucket_key == k.key) {
-			bucket_value = d.bucket_value;
-			break;
-		}
-	}
-	return bucket_value;
-}
-
 bool DataBucket::DeleteData(const DataBucketKey &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);
+				}
+
+				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()
+	);
+
 	return DataBucketsRepository::DeleteWhere(
 		database,
 		fmt::format(
@@ -170,16 +296,32 @@ bool DataBucket::DeleteData(const DataBucketKey &k)

 std::string DataBucket::GetDataExpires(const DataBucketKey &k)
 {
+	LogDataBuckets(
+		"Getting bucket expiration key [{}] bot_id [{}] character_id [{}] npc_id [{}]",
+		k.key,
+		k.bot_id,
+		k.character_id,
+		k.npc_id
+	);
+
 	auto r = GetData(k);
 	if (r.id == 0) {
 		return {};
 	}

-	return fmt::format("{}", r.expires);
+	return std::to_string(r.expires);
 }

 std::string DataBucket::GetDataRemaining(const DataBucketKey &k)
 {
+	LogDataBuckets(
+		"Getting bucket remaining key [{}] bot_id [{}] character_id [{}] npc_id [{}]",
+		k.key,
+		k.bot_id,
+		k.character_id,
+		k.npc_id
+	);
+
 	auto r = GetData(k);
 	if (r.id == 0) {
 		return "0";
@@ -218,3 +360,306 @@ std::string DataBucket::GetScopedDbFilters(const DataBucketKey &k)
 		!query.empty() ? "AND" : ""
 	);
 }
+
+bool DataBucket::CheckBucketMatch(const DataBucketsRepository::DataBuckets &dbe, const DataBucketKey &k)
+{
+	return (
+		dbe.key_ == k.key &&
+		dbe.bot_id == k.bot_id &&
+		dbe.character_id == k.character_id &&
+		dbe.npc_id == k.npc_id
+	);
+}
+
+void DataBucket::BulkLoadEntities(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) {
+			if (t == DataBucketLoadType::Bot) {
+				has_cache = ce.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];
+			}
+		}
+
+		if (has_cache) {
+			LogDataBucketsDetail("LoadType [{}] ID [{}] has cache", DataBucketLoadType::Name[t], ids[0]);
+			return;
+		}
+	}
+
+	std::string column;
+
+	switch (t) {
+		case DataBucketLoadType::Bot:
+			column = "bot_id";
+			break;
+		case DataBucketLoadType::Client:
+			column = "character_id";
+			break;
+		case DataBucketLoadType::NPC:
+			column = "npc_id";
+			break;
+		default:
+			LogError("Incorrect LoadType [{}]", t);
+			break;
+	}
+
+	const auto &l = DataBucketsRepository::GetWhere(
+		database,
+		fmt::format(
+			"{} IN ({}) AND (`expires` > {} OR `expires` = 0)",
+			column,
+			Strings::Join(ids, ", "),
+			(long long) std::time(nullptr)
+		)
+	);
+
+	if (l.empty()) {
+		return;
+	}
+
+	size_t size_before = g_data_bucket_cache.size();
+
+	LogDataBucketsDetail("cache size before [{}] l size [{}]", g_data_bucket_cache.size(), l.size());
+
+	uint32 added_count = 0;
+
+	for (const auto &e: l) {
+		if (!ExistsInCache(e)) {
+			added_count++;
+		}
+	}
+
+	g_data_bucket_cache.reserve(g_data_bucket_cache.size() + added_count);
+
+	for (const auto &e: l) {
+		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()
+				}
+			);
+		}
+	}
+
+	LogDataBucketsDetail("cache size after [{}]", g_data_bucket_cache.size());
+
+	LogDataBuckets(
+		"Bulk Loaded ids [{}] column [{}] new cache size is [{}]",
+		ids.size(),
+		column,
+		g_data_bucket_cache.size()
+	);
+}
+
+void DataBucket::DeleteCachedBuckets(DataBucketLoadType::Type t, uint32 id)
+{
+	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(),
+			[&](DataBucketCacheEntry &ce) {
+				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)
+				);
+			}
+		),
+		g_data_bucket_cache.end()
+	);
+
+	LogDataBuckets(
+		"LoadType [{}] id [{}] cache size before [{}] after [{}]",
+		DataBucketLoadType::Name[t],
+		id,
+		size_before,
+		g_data_bucket_cache.size()
+	);
+}
+
+int64_t DataBucket::GetCurrentTimeUNIX()
+{
+	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) {
+			return true;
+		}
+	}
+
+	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,
+		n.update_action
+	);
+
+	// delete
+	if (n.update_action == DataBucketCacheUpdateAction::Delete) {
+		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) {
+		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;
+		}
+
+		// update cache
+		if (ce.e.id == n.e.id) {
+			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
+	// this is to prevent the cache from growing too large
+	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(),
+			[&](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;
+			}
+		),
+		g_data_bucket_cache.end()
+	);
+	LogDataBucketsDetail(
+		"Deleted bucket misses from cache where key [{}] size before [{}] after [{}]",
+		e.key_,
+		size_before,
+		g_data_bucket_cache.size()
+	);
+}