|
|
@@ -1,6 +1,7 @@
|
|
|
#include "memory_store.hpp"
|
|
|
|
|
|
#include "config/collection_config_manager.hpp"
|
|
|
+#include "persistence/history_store.hpp"
|
|
|
|
|
|
#include <spdlog/spdlog.h>
|
|
|
|
|
|
@@ -120,15 +121,12 @@ bool MemoryStore::dropCollection(const std::string& name) {
|
|
|
{
|
|
|
std::shared_lock<std::shared_mutex> collLock(it->second->mutex);
|
|
|
docCount = it->second->documents.size();
|
|
|
- // Calculate total memory being freed (documents + version history + vectors)
|
|
|
+ // v1.9.0 — only live docs + vectors count toward the in-heap
|
|
|
+ // budget. Version history lives on disk now (see HistoryStore)
|
|
|
+ // and is reclaimed below via `historyStore_->dropCollection`.
|
|
|
for (const auto& [id, doc] : it->second->documents) {
|
|
|
totalSize += estimateDocumentSize(doc);
|
|
|
}
|
|
|
- for (const auto& [id, history] : it->second->versionHistory) {
|
|
|
- for (const auto& ver : history) {
|
|
|
- totalSize += estimateDocumentVersionSize(ver);
|
|
|
- }
|
|
|
- }
|
|
|
for (const auto& [id, vec] : it->second->vectors) {
|
|
|
totalSize += estimateVectorSize(vec);
|
|
|
}
|
|
|
@@ -136,6 +134,11 @@ bool MemoryStore::dropCollection(const std::string& name) {
|
|
|
|
|
|
collections_.erase(it);
|
|
|
|
|
|
+ // v1.9.0 — also drop the disk-resident history file for this collection.
|
|
|
+ if (historyStore_) {
|
|
|
+ historyStore_->dropCollection(name);
|
|
|
+ }
|
|
|
+
|
|
|
{
|
|
|
std::lock_guard<std::mutex> statsLock(statsMutex_);
|
|
|
stats_.totalCollections--;
|
|
|
@@ -166,24 +169,16 @@ bool MemoryStore::alterCollection(const std::string& name, const CollectionOptio
|
|
|
it->second->options = options;
|
|
|
it->second->updatedAt = currentTimeMs();
|
|
|
|
|
|
- // If maxVersions is now limited (and was unlimited or higher before), prune all history
|
|
|
- if (newMaxVersions > 0 && (oldMaxVersions == 0 || newMaxVersions < oldMaxVersions)) {
|
|
|
- uint64_t prunedVersions = 0;
|
|
|
- uint64_t prunedBytes = 0;
|
|
|
- for (auto& [docId, history] : it->second->versionHistory) {
|
|
|
- while (history.size() > newMaxVersions) {
|
|
|
- prunedBytes += estimateDocumentVersionSize(history.front());
|
|
|
- history.pop_front();
|
|
|
- ++prunedVersions;
|
|
|
- }
|
|
|
- }
|
|
|
- if (prunedBytes > 0) {
|
|
|
- estimatedMemoryBytes_.fetch_sub(prunedBytes, std::memory_order_relaxed);
|
|
|
- }
|
|
|
- if (prunedVersions > 0) {
|
|
|
- spdlog::info("Pruned {} version history entries from collection '{}' (maxVersions: {} -> {})",
|
|
|
- prunedVersions, name, oldMaxVersions, newMaxVersions);
|
|
|
- }
|
|
|
+ // v1.9.0 — if maxVersions is being lowered, rewrite the on-disk history
|
|
|
+ // file keeping at most newMaxVersions per docId. Pre-v1.9 this iterated
|
|
|
+ // the in-memory deques; the new HistoryStore::prune does the same logic
|
|
|
+ // on the .hlog file. Synchronous because alter is rare and operators
|
|
|
+ // expect their setting to take effect immediately.
|
|
|
+ if (historyStore_ && newMaxVersions > 0 &&
|
|
|
+ (oldMaxVersions == 0 || newMaxVersions < oldMaxVersions)) {
|
|
|
+ historyStore_->prune(name, newMaxVersions);
|
|
|
+ spdlog::info("Pruned on-disk version history for collection '{}' (maxVersions: {} -> {})",
|
|
|
+ name, oldMaxVersions, newMaxVersions);
|
|
|
}
|
|
|
|
|
|
spdlog::debug("Altered collection '{}': pinned={}, encrypted={}, maxVersions={}",
|
|
|
@@ -1611,6 +1606,14 @@ MemoryStore::MemoryStatsSnapshot MemoryStore::getMemoryStatsSnapshot() const {
|
|
|
// ===== Version History =====
|
|
|
|
|
|
void MemoryStore::saveToHistory(CollectionData& coll, const Document& currentDoc) {
|
|
|
+ // v1.9.0 — history is disk-resident now (see HistoryStore). If no
|
|
|
+ // historyStore_ is attached (tests, ad-hoc embeds), the call is a
|
|
|
+ // no-op — history is discarded. Pre-v1.9 this method built a
|
|
|
+ // DocumentVersion, pushed it onto an in-heap deque, and stamped
|
|
|
+ // `estimatedMemoryBytes_` with the version's overhead. That was
|
|
|
+ // the dominant source of the Zoe RSS gap; see BUG-memory-leak-zoe.md.
|
|
|
+ if (!historyStore_) return;
|
|
|
+
|
|
|
DocumentVersion ver;
|
|
|
ver.version = currentDoc.version;
|
|
|
ver.data = currentDoc.data;
|
|
|
@@ -1621,20 +1624,12 @@ void MemoryStore::saveToHistory(CollectionData& coll, const Document& currentDoc
|
|
|
ver.createdAt = currentDoc.createdAt;
|
|
|
ver.createdBy = currentDoc.createdBy;
|
|
|
|
|
|
- uint64_t addedBytes = estimateDocumentVersionSize(ver);
|
|
|
-
|
|
|
- auto& history = coll.versionHistory[currentDoc.id];
|
|
|
- history.push_back(std::move(ver));
|
|
|
- estimatedMemoryBytes_.fetch_add(addedBytes, std::memory_order_relaxed);
|
|
|
-
|
|
|
- // Prune oldest versions if maxVersions is configured
|
|
|
- if (coll.options.maxVersions > 0) {
|
|
|
- while (history.size() > coll.options.maxVersions) {
|
|
|
- uint64_t freed = estimateDocumentVersionSize(history.front());
|
|
|
- history.pop_front();
|
|
|
- estimatedMemoryBytes_.fetch_sub(freed, std::memory_order_relaxed);
|
|
|
- }
|
|
|
- }
|
|
|
+ historyStore_->append(currentDoc.collection, currentDoc.id, ver);
|
|
|
+ // Per-collection `max_versions` pruning runs in the snapshot loop
|
|
|
+ // (PersistenceManager::createSnapshot calls historyStore_->prune for
|
|
|
+ // every collection that has a cap set). Doing it inline on every
|
|
|
+ // write would either rewrite the file each time (expensive) or
|
|
|
+ // require an in-memory counter we no longer keep.
|
|
|
}
|
|
|
|
|
|
MemoryStore::VersionHistoryResult MemoryStore::getVersionHistory(
|
|
|
@@ -1656,34 +1651,21 @@ MemoryStore::VersionHistoryResult MemoryStore::getVersionHistory(
|
|
|
result.currentVersion = docIt->second.version;
|
|
|
}
|
|
|
|
|
|
- // Get version history
|
|
|
- auto histIt = coll->versionHistory.find(id);
|
|
|
- if (histIt == coll->versionHistory.end() || histIt->second.empty()) {
|
|
|
- // No history — still return current version info if document exists
|
|
|
- if (docIt == coll->documents.end()) {
|
|
|
- return result; // No document, no history
|
|
|
- }
|
|
|
- return result;
|
|
|
+ // v1.9.0 — historical versions come from the disk-resident
|
|
|
+ // HistoryStore. Newest-first ordering is preserved (HistoryStore
|
|
|
+ // returns them in that order). Without a HistoryStore attached the
|
|
|
+ // result is "no historical versions, current is whatever it is" —
|
|
|
+ // matching the empty-deque branch of the pre-v1.9 code.
|
|
|
+ if (historyStore_) {
|
|
|
+ result.totalCount = historyStore_->countVersions(collection, id);
|
|
|
+ result.versions = historyStore_->getVersions(collection, id, limit, offset);
|
|
|
}
|
|
|
|
|
|
- const auto& history = histIt->second;
|
|
|
- result.totalCount = history.size();
|
|
|
-
|
|
|
// Check if document was deleted (has history but no active document)
|
|
|
- if (docIt == coll->documents.end()) {
|
|
|
+ if (result.totalCount > 0 && docIt == coll->documents.end()) {
|
|
|
result.documentDeleted = true;
|
|
|
}
|
|
|
|
|
|
- // Apply pagination (newest first: offset=0 returns most recent versions)
|
|
|
- uint32_t total = static_cast<uint32_t>(history.size());
|
|
|
- uint32_t skipFromEnd = std::min(offset, total);
|
|
|
- uint32_t available = total - skipFromEnd;
|
|
|
- uint32_t count = limit > 0 ? std::min(limit, available) : available;
|
|
|
-
|
|
|
- for (uint32_t i = 0; i < count; ++i) {
|
|
|
- result.versions.push_back(history[total - skipFromEnd - 1 - i]);
|
|
|
- }
|
|
|
-
|
|
|
return result;
|
|
|
}
|
|
|
|
|
|
@@ -1713,19 +1695,9 @@ std::optional<DocumentVersion> MemoryStore::getDocumentAtVersion(
|
|
|
return ver;
|
|
|
}
|
|
|
|
|
|
- // Search in history
|
|
|
- auto histIt = coll->versionHistory.find(id);
|
|
|
- if (histIt == coll->versionHistory.end()) {
|
|
|
- return std::nullopt;
|
|
|
- }
|
|
|
-
|
|
|
- for (const auto& ver : histIt->second) {
|
|
|
- if (ver.version == version) {
|
|
|
- return ver;
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- return std::nullopt;
|
|
|
+ // v1.9.0 — historical versions live on disk now.
|
|
|
+ if (!historyStore_) return std::nullopt;
|
|
|
+ return historyStore_->getAtVersion(collection, id, version);
|
|
|
}
|
|
|
|
|
|
uint64_t MemoryStore::restoreToVersion(const std::string& collection, const std::string& id,
|
|
|
@@ -1734,23 +1706,14 @@ uint64_t MemoryStore::restoreToVersion(const std::string& collection, const std:
|
|
|
|
|
|
std::unique_lock<std::shared_mutex> lock(coll->mutex);
|
|
|
|
|
|
- // Find the target version in history
|
|
|
- auto histIt = coll->versionHistory.find(id);
|
|
|
- if (histIt == coll->versionHistory.end()) {
|
|
|
- return 0;
|
|
|
- }
|
|
|
-
|
|
|
- const DocumentVersion* targetVer = nullptr;
|
|
|
- for (const auto& ver : histIt->second) {
|
|
|
- if (ver.version == version) {
|
|
|
- targetVer = &ver;
|
|
|
- break;
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- if (!targetVer) {
|
|
|
- return 0;
|
|
|
- }
|
|
|
+ // v1.9.0 — pull the target version from disk-resident history. If
|
|
|
+ // no history store is wired (tests/embedded), restore is a no-op
|
|
|
+ // for non-current versions — matches pre-v1.9 behavior on an empty
|
|
|
+ // versionHistory map.
|
|
|
+ if (!historyStore_) return 0;
|
|
|
+ auto targetOpt = historyStore_->getAtVersion(collection, id, version);
|
|
|
+ if (!targetOpt) return 0;
|
|
|
+ const DocumentVersion& targetVer = *targetOpt;
|
|
|
|
|
|
auto docIt = coll->documents.find(id);
|
|
|
bool wasDeleted = (docIt == coll->documents.end());
|
|
|
@@ -1762,34 +1725,36 @@ uint64_t MemoryStore::restoreToVersion(const std::string& collection, const std:
|
|
|
saveToHistory(*coll, docIt->second);
|
|
|
|
|
|
newVersion = docIt->second.version + 1;
|
|
|
- docIt->second.data = targetVer->data;
|
|
|
+ docIt->second.data = targetVer.data;
|
|
|
docIt->second.version = newVersion;
|
|
|
docIt->second.updatedAt = currentTimeFor(collection);
|
|
|
docIt->second.updatedBy = actor;
|
|
|
- docIt->second.encrypted = targetVer->encrypted;
|
|
|
- docIt->second.encryptedFields = targetVer->encryptedFields;
|
|
|
+ docIt->second.encrypted = targetVer.encrypted;
|
|
|
+ docIt->second.encryptedFields = targetVer.encryptedFields;
|
|
|
docIt->second.nodeId = config_.nodeId;
|
|
|
|
|
|
restoredDoc = docIt->second;
|
|
|
} else {
|
|
|
- // Deleted document: reconstruct from version history
|
|
|
- // Find the highest version in history to determine new version number
|
|
|
- uint64_t maxVer = 0;
|
|
|
- for (const auto& ver : histIt->second) {
|
|
|
- maxVer = std::max(maxVer, ver.version);
|
|
|
+ // Deleted document: reconstruct from version history. Find the
|
|
|
+ // highest seen version to assign the next version number. With
|
|
|
+ // disk-resident history this is O(N) over the file, but the
|
|
|
+ // restore path is interactive / rare.
|
|
|
+ uint64_t maxVer = version; // at least the version we're restoring
|
|
|
+ for (const auto& v : historyStore_->getVersions(collection, id, 0, 0)) {
|
|
|
+ maxVer = std::max(maxVer, v.version);
|
|
|
}
|
|
|
newVersion = maxVer + 1;
|
|
|
|
|
|
restoredDoc.id = id;
|
|
|
restoredDoc.collection = collection;
|
|
|
- restoredDoc.data = targetVer->data;
|
|
|
+ restoredDoc.data = targetVer.data;
|
|
|
restoredDoc.version = newVersion;
|
|
|
- restoredDoc.createdAt = targetVer->createdAt;
|
|
|
- restoredDoc.createdBy = targetVer->createdBy;
|
|
|
+ restoredDoc.createdAt = targetVer.createdAt;
|
|
|
+ restoredDoc.createdBy = targetVer.createdBy;
|
|
|
restoredDoc.updatedAt = currentTimeFor(collection);
|
|
|
restoredDoc.updatedBy = actor;
|
|
|
- restoredDoc.encrypted = targetVer->encrypted;
|
|
|
- restoredDoc.encryptedFields = targetVer->encryptedFields;
|
|
|
+ restoredDoc.encrypted = targetVer.encrypted;
|
|
|
+ restoredDoc.encryptedFields = targetVer.encryptedFields;
|
|
|
restoredDoc.nodeId = config_.nodeId;
|
|
|
|
|
|
// Apply default TTL if configured
|
|
|
@@ -1838,14 +1803,15 @@ std::pair<uint64_t, uint64_t> MemoryStore::restoreToDate(
|
|
|
{
|
|
|
std::shared_lock<std::shared_mutex> lock(coll->mutex);
|
|
|
|
|
|
- // Search version history for the version active at the given timestamp
|
|
|
- auto histIt = coll->versionHistory.find(id);
|
|
|
- if (histIt != coll->versionHistory.end()) {
|
|
|
- for (const auto& ver : histIt->second) {
|
|
|
+ // v1.9.0 — historical versions are on disk. The HistoryStore
|
|
|
+ // returns them newest-first; we scan all of them to find the
|
|
|
+ // largest version whose timestamp is still <= the target.
|
|
|
+ if (historyStore_) {
|
|
|
+ for (const auto& ver : historyStore_->getVersions(collection, id, 0, 0)) {
|
|
|
if (ver.timestamp <= timestamp) {
|
|
|
+ // newest-first iteration: first match is the one we want
|
|
|
targetVersion = ver.version;
|
|
|
- } else {
|
|
|
- break; // Versions are ordered oldest-first
|
|
|
+ break;
|
|
|
}
|
|
|
}
|
|
|
}
|
|
|
@@ -2006,50 +1972,28 @@ void MemoryStore::loadDocumentWithHistory(const std::string& collection, Documen
|
|
|
}
|
|
|
|
|
|
std::vector<std::pair<std::string, std::deque<DocumentVersion>>>
|
|
|
-MemoryStore::getAllVersionHistory(const std::string& collection) const {
|
|
|
- const CollectionData* coll = getCollection(collection);
|
|
|
- if (!coll) {
|
|
|
- return {};
|
|
|
- }
|
|
|
-
|
|
|
- std::shared_lock<std::shared_mutex> lock(coll->mutex);
|
|
|
-
|
|
|
- std::vector<std::pair<std::string, std::deque<DocumentVersion>>> result;
|
|
|
- result.reserve(coll->versionHistory.size());
|
|
|
- for (const auto& [docId, history] : coll->versionHistory) {
|
|
|
- if (!history.empty()) {
|
|
|
- result.emplace_back(docId, history);
|
|
|
- }
|
|
|
- }
|
|
|
- return result;
|
|
|
+MemoryStore::getAllVersionHistory(const std::string& /*collection*/) const {
|
|
|
+ // v1.9.0 — history is on disk now; the snapshot writer is no longer
|
|
|
+ // responsible for it (see SnapshotManager::serializeStore which
|
|
|
+ // emits a zeroed history block for forward-compat). Returning an
|
|
|
+ // empty vector is the correct steady-state behavior. Kept on the
|
|
|
+ // class for API compat; no callers within the codebase use this
|
|
|
+ // for anything but snapshotting now.
|
|
|
+ return {};
|
|
|
}
|
|
|
|
|
|
void MemoryStore::loadVersionHistory(const std::string& collection, const std::string& docId,
|
|
|
std::deque<DocumentVersion> history) {
|
|
|
- CollectionData* coll = getOrCreateCollection(collection);
|
|
|
-
|
|
|
- std::unique_lock<std::shared_mutex> lock(coll->mutex);
|
|
|
-
|
|
|
- // Account for any history being replaced
|
|
|
- uint64_t oldBytes = 0;
|
|
|
- auto existing = coll->versionHistory.find(docId);
|
|
|
- if (existing != coll->versionHistory.end()) {
|
|
|
- for (const auto& ver : existing->second) {
|
|
|
- oldBytes += estimateDocumentVersionSize(ver);
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- uint64_t newBytes = 0;
|
|
|
- for (const auto& ver : history) {
|
|
|
- newBytes += estimateDocumentVersionSize(ver);
|
|
|
- }
|
|
|
-
|
|
|
- coll->versionHistory[docId] = std::move(history);
|
|
|
-
|
|
|
- if (newBytes > oldBytes) {
|
|
|
- estimatedMemoryBytes_.fetch_add(newBytes - oldBytes, std::memory_order_relaxed);
|
|
|
- } else if (oldBytes > newBytes) {
|
|
|
- estimatedMemoryBytes_.fetch_sub(oldBytes - newBytes, std::memory_order_relaxed);
|
|
|
+ // v1.9.0 — called by SnapshotManager when deserializing v1.8.x
|
|
|
+ // snapshots that still carried embedded history blocks. We migrate
|
|
|
+ // each version straight to the on-disk HistoryStore so operators
|
|
|
+ // don't lose retained history during the upgrade. If no
|
|
|
+ // HistoryStore is attached (tests), the call is a no-op — the
|
|
|
+ // history is silently dropped, matching "history feature off"
|
|
|
+ // semantics. Pre-v1.9 this method built the in-memory deque.
|
|
|
+ if (!historyStore_) return;
|
|
|
+ for (auto& ver : history) {
|
|
|
+ historyStore_->append(collection, docId, ver);
|
|
|
}
|
|
|
}
|
|
|
|
|
|
@@ -2879,23 +2823,10 @@ bool MemoryStore::evictDocument(const std::string& collection, const std::string
|
|
|
// Remove document from memory
|
|
|
coll->documents.erase(it);
|
|
|
|
|
|
- // v1.8.0 — also drop the doc's version history. Before this, eviction only
|
|
|
- // freed the live document but left `versionHistory[id]` intact, which on
|
|
|
- // hot-write collections with `max_versions: 0` (e.g. ShadowMan `messages`,
|
|
|
- // `documents`, `kb_*`) caused unbounded growth: every update pushed a
|
|
|
- // copy of the prior version into the deque, eviction never touched it,
|
|
|
- // and the orphaned deque stayed pinned to the heap forever. See
|
|
|
- // BUG-memory-leak-zoe.md. Hot-load via `loadEvictedDocument` already
|
|
|
- // doesn't restore history, so dropping it here is a no-op vs the
|
|
|
- // pre-v1.8 user-observable contract.
|
|
|
- uint64_t historyBytesFreed = 0;
|
|
|
- auto histIt = coll->versionHistory.find(id);
|
|
|
- if (histIt != coll->versionHistory.end()) {
|
|
|
- for (const auto& ver : histIt->second) {
|
|
|
- historyBytesFreed += estimateDocumentVersionSize(ver);
|
|
|
- }
|
|
|
- coll->versionHistory.erase(histIt);
|
|
|
- }
|
|
|
+ // v1.9.0 — version history is on disk now (HistoryStore), so eviction
|
|
|
+ // doesn't have to clean it up. The v1.8.0 in-memory history-prune
|
|
|
+ // code (which used to fire here) is gone; the unbounded-history
|
|
|
+ // problem it solved has been replaced with the on-disk model.
|
|
|
|
|
|
collLock.unlock();
|
|
|
|
|
|
@@ -2913,12 +2844,10 @@ bool MemoryStore::evictDocument(const std::string& collection, const std::string
|
|
|
stats_.evictionCount++;
|
|
|
}
|
|
|
|
|
|
- // Update memory tracking atomically (subtract evicted document + history)
|
|
|
- estimatedMemoryBytes_.fetch_sub(stub.estimatedSize + historyBytesFreed,
|
|
|
- std::memory_order_relaxed);
|
|
|
+ // Update memory tracking atomically (subtract evicted document size).
|
|
|
+ estimatedMemoryBytes_.fetch_sub(stub.estimatedSize, std::memory_order_relaxed);
|
|
|
|
|
|
- SPDLOG_DEBUG("Evicted document {}/{} (doc: {} bytes, history: {} bytes)",
|
|
|
- collection, id, stub.estimatedSize, historyBytesFreed);
|
|
|
+ SPDLOG_DEBUG("Evicted document {}/{} (size: {} bytes)", collection, id, stub.estimatedSize);
|
|
|
|
|
|
return true;
|
|
|
}
|