From 44ee6da6a0db9b0a4d3f2331c8654b052314cc04 Mon Sep 17 00:00:00 2001 From: CCG Date: Wed, 22 Jul 2026 11:57:49 -0300 Subject: [PATCH] Fix RemoteDB concurrent logic --- config/das.json | 51 +- src/atomdb/inmemorydb/InMemoryDB.cc | 182 +++--- src/atomdb/inmemorydb/InMemoryDB.h | 15 + src/atomdb/redis_mongodb/RedisMongoDB.cc | 10 +- src/atomdb/remotedb/RemoteAtomDB.cc | 93 +-- src/atomdb/remotedb/RemoteAtomDB.h | 4 +- src/atomdb/remotedb/RemoteAtomDBPeer.cc | 653 +++++++++++++------ src/atomdb/remotedb/RemoteAtomDBPeer.h | 37 +- src/tests/assets/remotedb_config_single.json | 4 + src/tests/cpp/BUILD | 2 +- src/tests/cpp/inmemorydb_test.cc | 124 ++++ src/tests/cpp/redis_mongodb_test.cc | 39 ++ src/tests/cpp/remote_atomdb_test.cc | 418 +++++++++--- src/tests/main/evaluation_evolution.cc | 22 +- 14 files changed, 1207 insertions(+), 447 deletions(-) diff --git a/config/das.json b/config/das.json index 4fb8241f..ae3de236 100644 --- a/config/das.json +++ b/config/das.json @@ -80,8 +80,7 @@ { "uid": "peer1", "type": "redismongodb", - "composite_type_enabled": true, - "context": "remotedb_test_peer1_", + "context": "", "mongodb": { "endpoint": "localhost:40021", "username": "admin", @@ -91,28 +90,48 @@ "endpoint": "localhost:40020", "cluster": false }, + "morkdb": { + "endpoint": "localhost:40022" + } + }, + { + "uid": "peer2", + "type": "redismongodb", + "context": "", + "mongodb": { + "endpoint": "localhost:40031", + "username": "admin", + "password": "admin" + }, + "redis": { + "endpoint": "localhost:40030", + "cluster": false + }, + "morkdb": { + "endpoint": "localhost:40032" + } + }, + { + "uid": "peer3", + "type": "inmemorydb", + "context": "", "local_persistence": { - "type": "morkdb", - "composite_type_enabled": true, - "context": "remotedb_test_peer1_local_", + "type": "redismongodb", + "context": "", + "composite_type_enabled": false, "mongodb": { - "endpoint": "localhost:40021", + "endpoint": "localhost:40041", "username": "admin", "password": "admin" }, + "redis": { + "endpoint": "localhost:40040", + "cluster": false + }, "morkdb": { - "endpoint": "localhost:40022" + "endpoint": "localhost:40042" } } - }, - { - "uid": "peer2", - "type": "inmemorydb", - "context": "remotedb_test_peer2_", - "local_persistence": { - "type": "inmemorydb", - "context": "remotedb_test_peer2_local_" - } } ] }, diff --git a/src/atomdb/inmemorydb/InMemoryDB.cc b/src/atomdb/inmemorydb/InMemoryDB.cc index edcad1a4..f3cf5616 100644 --- a/src/atomdb/inmemorydb/InMemoryDB.cc +++ b/src/atomdb/inmemorydb/InMemoryDB.cc @@ -62,6 +62,25 @@ struct ReIndexData { InMemoryDB* db; }; +shared_ptr clone_atom(const Atom* atom) { + if (atom == nullptr) { + return nullptr; + } + + if (atom->arity() == 0) { + auto node = dynamic_cast(atom); + return (node != nullptr) ? make_shared(*node) : nullptr; + } + + auto link = dynamic_cast(atom); + return (link != nullptr) ? make_shared(*link) : nullptr; +} + +void reset_trie(HandleTrie*& trie) { + delete trie; + trie = new HandleTrie(HANDLE_HASH_SIZE - 1); +} + bool re_index_visitor(HandleTrie::TrieNode* node, void* data) { ReIndexData* index_data = static_cast(data); if (node->value != NULL) { @@ -90,49 +109,15 @@ InMemoryDB::InMemoryDB(const string& context) incoming_sets_trie_(new HandleTrie(HANDLE_HASH_SIZE - 1)) {} InMemoryDB::~InMemoryDB() { - // Traverse and delete all atoms - this->atoms_trie_->traverse( - false, - [](HandleTrie::TrieNode* node, void* data) -> bool { - if (node->value != NULL) { - delete node->value; - node->value = NULL; - } - return false; // Continue traversal - }, - NULL); delete this->atoms_trie_; - - // Traverse and delete all pattern index entries - this->pattern_index_trie_->traverse( - false, - [](HandleTrie::TrieNode* node, void* data) -> bool { - if (node->value != NULL) { - delete node->value; - node->value = NULL; - } - return false; // Continue traversal - }, - NULL); delete this->pattern_index_trie_; - - // Traverse and delete all incoming set entries - this->incoming_sets_trie_->traverse( - false, - [](HandleTrie::TrieNode* node, void* data) -> bool { - if (node->value != NULL) { - delete node->value; - node->value = NULL; - } - return false; // Continue traversal - }, - NULL); delete this->incoming_sets_trie_; } bool InMemoryDB::allow_nested_indexing() { return false; } shared_ptr InMemoryDB::get_atom(const string& handle) { + lock_guard lock(api_mutex_); auto trie_value = this->atoms_trie_->lookup(handle); if (trie_value == NULL) { return nullptr; @@ -141,34 +126,22 @@ shared_ptr InMemoryDB::get_atom(const string& handle) { if (atom_trie_value == NULL) { return nullptr; } - // Clone the atom to return a shared_ptr (caller doesn't own the original) - Atom* atom = atom_trie_value->get_atom(); - if (atom->arity() == 0) { - auto node = dynamic_cast(atom); - return make_shared(*node); - } else { - auto link = dynamic_cast(atom); - return make_shared(*link); - } + // Return a deep copy (caller must not observe internal trie-owned storage). + return clone_atom(atom_trie_value->get_atom()); } shared_ptr InMemoryDB::get_node(const string& handle) { auto atom = get_atom(handle); - if (atom != nullptr) { - return make_shared(*dynamic_cast(atom.get())); - } - return nullptr; + return dynamic_pointer_cast(atom); } shared_ptr InMemoryDB::get_link(const string& handle) { auto atom = get_atom(handle); - if (atom != nullptr) { - return make_shared(*dynamic_cast(atom.get())); - } - return nullptr; + return dynamic_pointer_cast(atom); } shared_ptr InMemoryDB::query_for_pattern(const LinkSchema& link_schema) { + lock_guard lock(api_mutex_); auto handle_set = make_shared(); // Check if we have this pattern indexed in the HandleTrie @@ -184,6 +157,7 @@ shared_ptr InMemoryDB::query_for_pattern(const LinkSchema& link_schem } shared_ptr InMemoryDB::query_for_targets(const string& handle) { + lock_guard lock(api_mutex_); auto trie_value = atoms_trie_->lookup(handle); if (trie_value == NULL) { return nullptr; @@ -201,6 +175,7 @@ shared_ptr InMemoryDB::query_for_targets(const string& handle) { } shared_ptr InMemoryDB::query_for_incoming_set(const string& handle) { + lock_guard lock(api_mutex_); auto handle_set = make_shared(); auto incoming_set_trie_value = dynamic_cast(this->incoming_sets_trie_->lookup(handle)); @@ -213,6 +188,7 @@ shared_ptr InMemoryDB::query_for_incoming_set(const string& handle) { } vector> InMemoryDB::get_matching_atoms(bool is_toplevel, Atom& key) { + lock_guard lock(api_mutex_); vector> matching_atoms; auto trie_value = atoms_trie_->lookup(key.handle()); if (trie_value == NULL) { @@ -231,9 +207,13 @@ vector> InMemoryDB::get_matching_atoms(bool is_toplevel, Atom& return matching_atoms; } -bool InMemoryDB::atom_exists(const string& handle) { return atoms_trie_->lookup(handle) != NULL; } +bool InMemoryDB::atom_exists(const string& handle) { + lock_guard lock(api_mutex_); + return atoms_trie_->lookup(handle) != NULL; +} bool InMemoryDB::node_exists(const string& handle) { + lock_guard lock(api_mutex_); auto trie_value = atoms_trie_->lookup(handle); if (trie_value == NULL) { return false; @@ -247,6 +227,7 @@ bool InMemoryDB::node_exists(const string& handle) { } bool InMemoryDB::link_exists(const string& handle) { + lock_guard lock(api_mutex_); auto trie_value = atoms_trie_->lookup(handle); if (trie_value == NULL) { return false; @@ -260,6 +241,7 @@ bool InMemoryDB::link_exists(const string& handle) { } set InMemoryDB::atoms_exist(const vector& handles) { + lock_guard lock(api_mutex_); set existing; for (const auto& handle : handles) { if (atoms_trie_->lookup(handle) != NULL) { @@ -270,6 +252,7 @@ set InMemoryDB::atoms_exist(const vector& handles) { } set InMemoryDB::nodes_exist(const vector& handles) { + lock_guard lock(api_mutex_); set existing; for (const auto& handle : handles) { if (this->node_exists(handle)) { @@ -280,6 +263,7 @@ set InMemoryDB::nodes_exist(const vector& handles) { } set InMemoryDB::links_exist(const vector& handles) { + lock_guard lock(api_mutex_); set existing; for (const auto& handle : handles) { if (this->link_exists(handle)) { @@ -298,6 +282,7 @@ string InMemoryDB::add_atom(const atoms::Atom* atom, bool throw_if_exists) { } string InMemoryDB::add_node(const atoms::Node* node, bool throw_if_exists) { + lock_guard lock(api_mutex_); string handle = node->handle(); if (throw_if_exists && this->node_exists(handle)) { @@ -305,16 +290,9 @@ string InMemoryDB::add_node(const atoms::Node* node, bool throw_if_exists) { return ""; } - // Check if already exists - auto existing = atoms_trie_->lookup(handle); - if (existing != NULL && !throw_if_exists) { - return handle; // Already exists, return handle - } - - // Clone the node to store in trie + // Upsert: HandleTrie::insert merges via AtomTrieValue::merge (replaces the stored atom). Node* cloned_node = new Node(*node); - auto atom_trie_value = new AtomTrieValue(cloned_node); - atoms_trie_->insert(handle, atom_trie_value); + atoms_trie_->insert(handle, new AtomTrieValue(cloned_node)); return handle; } @@ -331,6 +309,8 @@ vector InMemoryDB::add_atoms(const vector& atoms, if (atoms.empty()) { return {}; } + // Lock the whole batch so it is atomic w.r.t. concurrent readers. + lock_guard lock(api_mutex_); vector nodes; vector links; @@ -355,6 +335,7 @@ vector InMemoryDB::add_nodes(const vector& nodes, if (nodes.empty()) { return {}; } + lock_guard lock(api_mutex_); vector handles; for (const auto& node : nodes) { @@ -384,6 +365,7 @@ vector InMemoryDB::add_links(const vector& links, if (links.empty()) { return {}; } + lock_guard lock(api_mutex_); if (throw_if_exists) { vector handles; @@ -404,22 +386,18 @@ vector InMemoryDB::add_links(const vector& links, string link_handle = link->handle(); handles.push_back(link_handle); - // Check if already exists - auto existing = atoms_trie_->lookup(link_handle); - if (existing == NULL || !throw_if_exists) { - if (existing == NULL) { - // Clone the link to store in trie - Link* cloned_link = new Link(*link); - auto atom_trie_value = new AtomTrieValue(cloned_link); - atoms_trie_->insert(link_handle, atom_trie_value); - } + bool is_new = atoms_trie_->lookup(link_handle) == NULL; + + // Upsert: insert merges via AtomTrieValue::merge when the handle already exists. + // Content-addressed handles share targets, so indexes only need building on first insert. + Link* cloned_link = new Link(*link); + atoms_trie_->insert(link_handle, new AtomTrieValue(cloned_link)); - // Update incoming sets for each target + if (is_new) { for (const auto& target_handle : link->targets) { this->add_incoming_set(target_handle, link_handle); } - // Index pattern auto pattern_handles = this->match_pattern_index_schema(link); for (const auto& pattern_handle : pattern_handles) { this->add_pattern(pattern_handle, link_handle); @@ -431,6 +409,7 @@ vector InMemoryDB::add_links(const vector& links, } bool InMemoryDB::delete_atom(const string& handle, bool delete_link_targets) { + lock_guard lock(api_mutex_); if (this->delete_node(handle, delete_link_targets)) { return true; } @@ -438,6 +417,7 @@ bool InMemoryDB::delete_atom(const string& handle, bool delete_link_targets) { } bool InMemoryDB::delete_node(const string& handle, bool delete_link_targets) { + lock_guard lock(api_mutex_); auto trie_value = this->atoms_trie_->lookup(handle); if (trie_value == NULL) { return false; @@ -480,6 +460,7 @@ bool InMemoryDB::delete_node(const string& handle, bool delete_link_targets) { } bool InMemoryDB::delete_link(const string& handle, bool delete_link_targets) { + lock_guard lock(api_mutex_); auto trie_value = atoms_trie_->lookup(handle); if (trie_value == NULL) { return false; @@ -532,6 +513,7 @@ bool InMemoryDB::delete_link(const string& handle, bool delete_link_targets) { } uint InMemoryDB::delete_atoms(const vector& handles, bool delete_link_targets) { + lock_guard lock(api_mutex_); uint deleted_count = 0; for (const auto& handle : handles) { if (this->delete_atom(handle, delete_link_targets)) { @@ -542,6 +524,7 @@ uint InMemoryDB::delete_atoms(const vector& handles, bool delete_link_ta } uint InMemoryDB::delete_nodes(const vector& handles, bool delete_link_targets) { + lock_guard lock(api_mutex_); uint deleted_count = 0; for (const auto& handle : handles) { if (this->delete_node(handle, delete_link_targets)) { @@ -552,6 +535,7 @@ uint InMemoryDB::delete_nodes(const vector& handles, bool delete_link_ta } uint InMemoryDB::delete_links(const vector& handles, bool delete_link_targets) { + lock_guard lock(api_mutex_); uint deleted_count = 0; for (const auto& handle : handles) { if (this->delete_link(handle, delete_link_targets)) { @@ -566,25 +550,48 @@ size_t InMemoryDB::node_count() const { RAISE_ERROR("node_count() is not impleme size_t InMemoryDB::link_count() const { RAISE_ERROR("link_count() is not implemented yet"); } size_t InMemoryDB::atom_count() const { + lock_guard lock(api_mutex_); auto size = this->atoms_trie_->size; return static_cast(size); } +vector> InMemoryDB::get_all_atoms() { + lock_guard lock(api_mutex_); + vector> atoms; + atoms.reserve(this->atoms_trie_->size); + this->atoms_trie_->traverse( + false, + [](HandleTrie::TrieNode* node, void* data) -> bool { + if (node->value == nullptr) { + return false; + } + auto atom_trie_value = dynamic_cast(node->value); + if (atom_trie_value == nullptr) { + return false; + } + auto* out = static_cast>*>(data); + auto cloned = clone_atom(atom_trie_value->get_atom()); + if (cloned != nullptr) { + out->push_back(cloned); + } + return false; + }, + &atoms); + return atoms; +} + +void InMemoryDB::drop_all() { + lock_guard lock(api_mutex_); + reset_trie(this->atoms_trie_); + reset_trie(this->pattern_index_trie_); + reset_trie(this->incoming_sets_trie_); +} + void InMemoryDB::re_index_patterns(bool flush_patterns) { + lock_guard lock(api_mutex_); if (flush_patterns) { - // Clear all pattern index entries by deleting and recreating the trie - this->pattern_index_trie_->traverse( - false, - [](HandleTrie::TrieNode* node, void* data) -> bool { - if (node->value != NULL) { - delete node->value; - node->value = NULL; - } - return false; // Continue traversal - }, - NULL); - delete this->pattern_index_trie_; - this->pattern_index_trie_ = new HandleTrie(HANDLE_HASH_SIZE - 1); + // Clear all pattern index entries by recreating the trie. + reset_trie(this->pattern_index_trie_); } // Re-index all links @@ -595,6 +602,7 @@ void InMemoryDB::re_index_patterns(bool flush_patterns) { // Helper methods void InMemoryDB::add_pattern(const string& pattern_handle, const string& atom_handle) { + lock_guard lock(api_mutex_); auto pattern_trie_value = dynamic_cast(this->pattern_index_trie_->lookup(pattern_handle)); if (pattern_trie_value == NULL) { @@ -652,6 +660,7 @@ void InMemoryDB::update_incoming_set(const string& target_handle, const string& void InMemoryDB::add_pattern_index_schema(const string& tokens, const vector>& index_entries) { + lock_guard lock(api_mutex_); auto tokens_vector = Utils::split(tokens, ' '); LinkSchema link_schema(tokens_vector); @@ -661,6 +670,7 @@ void InMemoryDB::add_pattern_index_schema(const string& tokens, } vector InMemoryDB::match_pattern_index_schema(const Link* link) { + lock_guard lock(api_mutex_); vector pattern_handles; const auto& map_ref = this->pattern_index_schema_map; diff --git a/src/atomdb/inmemorydb/InMemoryDB.h b/src/atomdb/inmemorydb/InMemoryDB.h index cc683d1c..e1ed99e0 100644 --- a/src/atomdb/inmemorydb/InMemoryDB.h +++ b/src/atomdb/inmemorydb/InMemoryDB.h @@ -1,6 +1,7 @@ #pragma once #include +#include #include #include #include @@ -16,6 +17,14 @@ using namespace atoms; namespace atomdb { +/** + * In-memory AtomDB backed by HandleTries. + * + * Thread-safety: all public methods are serialized by an internal recursive mutex, so an + * InMemoryDB instance can be shared across threads without external locking. The mutex is + * recursive because public methods call each other (e.g. delete_link -> delete_atom, and + * re_index_patterns' visitor calls back into add_pattern / match_pattern_index_schema). + */ class InMemoryDB : public AtomDB { public: InMemoryDB(const string& context = ""); @@ -72,8 +81,14 @@ class InMemoryDB : public AtomDB { void re_index_patterns(bool flush_patterns = true) override; + vector> get_all_atoms(); + void drop_all(); + private: string context_; + // Serializes all public methods (see class comment). Recursive: public methods re-enter + // each other and LinkSchema::match may call back into this AtomDB on the same thread. + mutable recursive_mutex api_mutex_; HandleTrie* atoms_trie_; // Stores handle -> Atom* HandleTrie* pattern_index_trie_; // Stores pattern_handle -> set of atom handles HandleTrie* incoming_sets_trie_; // Stores target_handle -> set of link handles that reference it diff --git a/src/atomdb/redis_mongodb/RedisMongoDB.cc b/src/atomdb/redis_mongodb/RedisMongoDB.cc index 8a9bc47b..ef653fc1 100644 --- a/src/atomdb/redis_mongodb/RedisMongoDB.cc +++ b/src/atomdb/redis_mongodb/RedisMongoDB.cc @@ -906,10 +906,12 @@ vector RedisMongoDB::add_links(const vector& links, } map> composite_type_entries_map; - if (this->composite_type_enabled_ && is_transactional) { - this->build_composite_type_entries_map(links, composite_type_entries_map); - } else if (!is_transactional) { - this->check_existing_targets(links); + if (this->composite_type_enabled_) { + if (is_transactional) { + this->build_composite_type_entries_map(links, composite_type_entries_map); + } else { + this->check_existing_targets(links); + } } vector handles; diff --git a/src/atomdb/remotedb/RemoteAtomDB.cc b/src/atomdb/remotedb/RemoteAtomDB.cc index 82d9569b..ea1ce9d2 100644 --- a/src/atomdb/remotedb/RemoteAtomDB.cc +++ b/src/atomdb/remotedb/RemoteAtomDB.cc @@ -1,6 +1,7 @@ #define LOG_LEVEL INFO_LEVEL #include "RemoteAtomDB.h" +#include #include #include #include @@ -9,8 +10,10 @@ #include "InMemoryDB.h" #include "InMemoryDBAPITypes.h" +#include "Link.h" #include "Logger.h" #include "MorkDB.h" +#include "Node.h" #include "RedisMongoDB.h" #include "Utils.h" @@ -113,13 +116,25 @@ void RemoteAtomDB::derive_nested_indexing() { bool RemoteAtomDB::allow_nested_indexing() { return nested_indexing_; } shared_ptr RemoteAtomDB::get_atom(const string& handle) { - // Phase 1: probe every peer's in-memory cache first (no network). Silent: this is the hot path. + // Writable peers first: local_persistence is the source of truth for updated custom + // attributes (strength) that share a content-addressed handle. for (auto& [uid, peer] : remote_db_) { + if (peer->is_readonly()) continue; + auto atom = peer->get_atom(handle); + if (atom) { + LOG_DEBUG("get_atom(" << handle << ") fetched from writable peer [" << uid << "]"); + return atom; + } + } + + // Readonly peers: cache probe then escalate to remote backends (base KB hot path). + for (auto& [uid, peer] : remote_db_) { + if (!peer->is_readonly()) continue; auto atom = peer->get_cached_atom(handle); if (atom) return atom; } - // Phase 2: escalate to peers (local_persistence + remote backend) only when no cache has it. for (auto& [uid, peer] : remote_db_) { + if (!peer->is_readonly()) continue; auto atom = peer->get_atom(handle); if (atom) { LOG_DEBUG("get_atom(" << handle << ") fetched from [" << uid << "]"); @@ -131,35 +146,13 @@ shared_ptr RemoteAtomDB::get_atom(const string& handle) { } shared_ptr RemoteAtomDB::get_node(const string& handle) { - for (auto& [uid, peer] : remote_db_) { - auto node = peer->get_cached_node(handle); - if (node) return node; - } - for (auto& [uid, peer] : remote_db_) { - auto node = peer->get_node(handle); - if (node) { - LOG_DEBUG("get_node(" << handle << ") fetched from [" << uid << "]"); - return node; - } - } - LOG_DEBUG("get_node(" << handle << ") not found in any peer"); - return nullptr; + auto atom = get_atom(handle); + return dynamic_pointer_cast(atom); } shared_ptr RemoteAtomDB::get_link(const string& handle) { - for (auto& [uid, peer] : remote_db_) { - auto link = peer->get_cached_link(handle); - if (link) return link; - } - for (auto& [uid, peer] : remote_db_) { - auto link = peer->get_link(handle); - if (link) { - LOG_DEBUG("get_link(" << handle << ") fetched from [" << uid << "]"); - return link; - } - } - LOG_DEBUG("get_link(" << handle << ") not found in any peer"); - return nullptr; + auto atom = get_atom(handle); + return dynamic_pointer_cast(atom); } vector> RemoteAtomDB::get_matching_atoms(bool is_toplevel, Atom& key) { @@ -326,10 +319,12 @@ set RemoteAtomDB::links_exist(const vector& handles) { } string RemoteAtomDB::add_atom(const atoms::Atom* atom, bool throw_if_exists) { + // Writes only land on writable peers (readonly peers gate the call internally). string handle; for (auto& [uid, peer] : remote_db_) { LOG_DEBUG("add_atom(" << atom->handle() << ") to peer [" << uid << "]"); - handle = peer->add_atom(atom, throw_if_exists); + string peer_handle = peer->add_atom(atom, throw_if_exists); + if (!peer_handle.empty()) handle = peer_handle; } return handle; } @@ -338,7 +333,8 @@ string RemoteAtomDB::add_node(const atoms::Node* node, bool throw_if_exists) { string handle; for (auto& [uid, peer] : remote_db_) { LOG_DEBUG("add_node(" << node->handle() << ") to peer [" << uid << "]"); - handle = peer->add_node(node, throw_if_exists); + string peer_handle = peer->add_node(node, throw_if_exists); + if (!peer_handle.empty()) handle = peer_handle; } return handle; } @@ -347,7 +343,8 @@ string RemoteAtomDB::add_link(const atoms::Link* link, bool throw_if_exists) { string handle; for (auto& [uid, peer] : remote_db_) { LOG_DEBUG("add_link(" << link->handle() << ") to peer [" << uid << "]"); - handle = peer->add_link(link, throw_if_exists); + string peer_handle = peer->add_link(link, throw_if_exists); + if (!peer_handle.empty()) handle = peer_handle; } return handle; } @@ -358,7 +355,8 @@ vector RemoteAtomDB::add_atoms(const vector& atoms, vector handles; for (auto& [uid, peer] : remote_db_) { LOG_DEBUG("add_atoms(" << atoms.size() << ") to peer [" << uid << "]"); - handles = peer->add_atoms(atoms, throw_if_exists, is_transactional); + auto peer_handles = peer->add_atoms(atoms, throw_if_exists, is_transactional); + if (!peer_handles.empty()) handles = peer_handles; } return handles; } @@ -369,7 +367,8 @@ vector RemoteAtomDB::add_nodes(const vector& nodes, vector handles; for (auto& [uid, peer] : remote_db_) { LOG_DEBUG("add_nodes(" << nodes.size() << ") to peer [" << uid << "]"); - handles = peer->add_nodes(nodes, throw_if_exists, is_transactional); + auto peer_handles = peer->add_nodes(nodes, throw_if_exists, is_transactional); + if (!peer_handles.empty()) handles = peer_handles; } return handles; } @@ -380,34 +379,35 @@ vector RemoteAtomDB::add_links(const vector& links, vector handles; for (auto& [uid, peer] : remote_db_) { LOG_DEBUG("add_links(" << links.size() << ") to peer [" << uid << "]"); - handles = peer->add_links(links, throw_if_exists, is_transactional); + auto peer_handles = peer->add_links(links, throw_if_exists, is_transactional); + if (!peer_handles.empty()) handles = peer_handles; } return handles; } bool RemoteAtomDB::delete_atom(const string& handle, bool delete_link_targets) { - bool ok = true; + bool ok = false; for (auto& [uid, peer] : remote_db_) { LOG_DEBUG("delete_atom(" << handle << ") from peer [" << uid << "]"); - ok = peer->delete_atom(handle, delete_link_targets) && ok; + ok = peer->delete_atom(handle, delete_link_targets) || ok; } return ok; } bool RemoteAtomDB::delete_node(const string& handle, bool delete_link_targets) { - bool ok = true; + bool ok = false; for (auto& [uid, peer] : remote_db_) { LOG_DEBUG("delete_node(" << handle << ") from peer [" << uid << "]"); - ok = peer->delete_node(handle, delete_link_targets) && ok; + ok = peer->delete_node(handle, delete_link_targets) || ok; } return ok; } bool RemoteAtomDB::delete_link(const string& handle, bool delete_link_targets) { - bool ok = true; + bool ok = false; for (auto& [uid, peer] : remote_db_) { LOG_DEBUG("delete_link(" << handle << ") from peer [" << uid << "]"); - ok = peer->delete_link(handle, delete_link_targets) && ok; + ok = peer->delete_link(handle, delete_link_targets) || ok; } return ok; } @@ -416,7 +416,7 @@ uint RemoteAtomDB::delete_atoms(const vector& handles, bool delete_link_ uint count = 0; for (auto& [uid, peer] : remote_db_) { LOG_DEBUG("delete_atoms(" << handles.size() << ") from peer [" << uid << "]"); - count = peer->delete_atoms(handles, delete_link_targets); + count = max(count, peer->delete_atoms(handles, delete_link_targets)); } return count; } @@ -425,7 +425,7 @@ uint RemoteAtomDB::delete_nodes(const vector& handles, bool delete_link_ uint count = 0; for (auto& [uid, peer] : remote_db_) { LOG_DEBUG("delete_nodes(" << handles.size() << ") from peer [" << uid << "]"); - count = peer->delete_nodes(handles, delete_link_targets); + count = max(count, peer->delete_nodes(handles, delete_link_targets)); } return count; } @@ -434,7 +434,7 @@ uint RemoteAtomDB::delete_links(const vector& handles, bool delete_link_ uint count = 0; for (auto& [uid, peer] : remote_db_) { LOG_DEBUG("delete_links(" << handles.size() << ") from peer [" << uid << "]"); - count = peer->delete_links(handles, delete_link_targets); + count = max(count, peer->delete_links(handles, delete_link_targets)); } return count; } @@ -474,3 +474,10 @@ RemoteAtomDBPeer* RemoteAtomDB::get_peer(const string& uid) { auto it = remote_db_.find(uid); return (it != remote_db_.end()) ? it->second.get() : nullptr; } + +void RemoteAtomDB::release_caches(const LinkSchema& link_schema, bool persist, bool force) { + for (auto& [uid, peer] : remote_db_) { + LOG_DEBUG("release_caches(" << link_schema.handle() << ") from peer [" << uid << "]"); + peer->release(link_schema, persist, force); + } +} diff --git a/src/atomdb/remotedb/RemoteAtomDB.h b/src/atomdb/remotedb/RemoteAtomDB.h index b63278a9..40b09ea7 100644 --- a/src/atomdb/remotedb/RemoteAtomDB.h +++ b/src/atomdb/remotedb/RemoteAtomDB.h @@ -14,7 +14,7 @@ namespace atomdb { /** * RemoteAtomDB connects to multiple remote AtomDBs via RemoteAtomDBPeer instances. - * Each peer maintains its own cache, remote connection, and local persistence. + * Each peer maintains its own optional cache, remote connection, and optional local persistence. * The constructor expects a JSON config with connection info for each remote peer. */ class RemoteAtomDB : public AtomDB { @@ -79,6 +79,8 @@ class RemoteAtomDB : public AtomDB { const map>& get_remote_dbs() const { return remote_db_; } RemoteAtomDBPeer* get_peer(const string& uid); + void release_caches(const LinkSchema& link_schema, bool persist = true, bool force = false); + private: // Derives the aggregated nested-indexing capability from the current peers. Shared by both // constructors so the config and DI paths stay consistent. diff --git a/src/atomdb/remotedb/RemoteAtomDBPeer.cc b/src/atomdb/remotedb/RemoteAtomDBPeer.cc index 4b68df11..6cf6c495 100644 --- a/src/atomdb/remotedb/RemoteAtomDBPeer.cc +++ b/src/atomdb/remotedb/RemoteAtomDBPeer.cc @@ -2,8 +2,10 @@ #include "RemoteAtomDBPeer.h" #include +#include #include #include +#include #include "Atom.h" #include "InMemoryDBAPITypes.h" @@ -22,112 +24,74 @@ RemoteAtomDBPeer::RemoteAtomDBPeer(shared_ptr remote_atomdb, shared_ptr local_persistence, const string& uid) : uid_(uid), - cache_(uid), + cache_(make_shared(uid)), atomdb_(remote_atomdb), local_persistence_(local_persistence), - fetched_link_templates_(HANDLE_HASH_SIZE - 1) { - empty_trie_value_ = new EmptyTrieValue(); + fetched_link_templates_(make_unique(HANDLE_HASH_SIZE - 1)) { start_cleanup_thread(); } RemoteAtomDBPeer::~RemoteAtomDBPeer() { stop_cleanup_thread(); } -bool RemoteAtomDBPeer::allow_nested_indexing() { - return atomdb_ ? atomdb_->allow_nested_indexing() : false; -} +bool RemoteAtomDBPeer::allow_nested_indexing() { return atomdb_->allow_nested_indexing(); } bool RemoteAtomDBPeer::composite_type_enabled() const { return local_persistence_ && local_persistence_->composite_type_enabled(); } -shared_ptr RemoteAtomDBPeer::get_atom(const string& handle) { - auto atom = cache_.get_atom(handle); - if (atom) return atom; +shared_ptr RemoteAtomDBPeer::cache() const { + lock_guard lock(peer_mutex_); + return cache_; +} +shared_ptr RemoteAtomDBPeer::get_atom(const string& handle) { + // local_persistence is authoritative for written/updated atoms (including strength changes + // that keep the same content-addressed handle). Prefer it over a possibly stale cache. if (local_persistence_) { - atom = local_persistence_->get_atom(handle); + auto atom = local_persistence_->get_atom(handle); if (atom) { - LOG_DEBUG("[" << uid_ << "] get_atom(" << handle << ") <- local_persistence (cached)"); - cache_.add_atom(atom.get()); + LOG_DEBUG("[RemoteDB(" << uid_ << ")] get_atom(" << handle + << ") <- local_persistence (cached)"); + cache()->add_atom(atom.get()); return atom; } } - if (atomdb_) { - atom = atomdb_->get_atom(handle); - if (atom) { - LOG_DEBUG("[" << uid_ << "] get_atom(" << handle << ") <- remote atomdb (cached)"); - cache_.add_atom(atom.get()); - return atom; - } + if (auto atom = cache()->get_atom(handle)) { + return atom; + } + + auto atom = atomdb_->get_atom(handle); + if (atom) { + LOG_DEBUG("[RemoteDB(" << uid_ << ")] get_atom(" << handle << ") <- remote atomdb (cached)"); + cache()->add_atom(atom.get()); + return atom; } - LOG_DEBUG("[" << uid_ << "] get_atom(" << handle << ") miss"); + LOG_DEBUG("[RemoteDB(" << uid_ << ")] get_atom(" << handle << ") miss"); return nullptr; } shared_ptr RemoteAtomDBPeer::get_node(const string& handle) { - auto node = cache_.get_node(handle); - if (node) return node; - - if (local_persistence_) { - node = local_persistence_->get_node(handle); - if (node) { - LOG_DEBUG("[" << uid_ << "] get_node(" << handle << ") <- local_persistence (cached)"); - cache_.add_node(node.get()); - return node; - } - } - - if (atomdb_) { - node = atomdb_->get_node(handle); - if (node) { - LOG_DEBUG("[" << uid_ << "] get_node(" << handle << ") <- remote atomdb (cached)"); - cache_.add_node(node.get()); - return node; - } - } - - LOG_DEBUG("[" << uid_ << "] get_node(" << handle << ") miss"); - return nullptr; + auto atom = get_atom(handle); + return dynamic_pointer_cast(atom); } shared_ptr RemoteAtomDBPeer::get_link(const string& handle) { - auto link = cache_.get_link(handle); - if (link) return link; - - if (local_persistence_) { - link = local_persistence_->get_link(handle); - if (link) { - LOG_DEBUG("[" << uid_ << "] get_link(" << handle << ") <- local_persistence (cached)"); - cache_.add_link(link.get()); - return link; - } - } - - if (atomdb_) { - link = atomdb_->get_link(handle); - if (link) { - LOG_DEBUG("[" << uid_ << "] get_link(" << handle << ") <- remote atomdb (cached)"); - cache_.add_link(link.get()); - return link; - } - } - - LOG_DEBUG("[" << uid_ << "] get_link(" << handle << ") miss"); - return nullptr; + auto atom = get_atom(handle); + return dynamic_pointer_cast(atom); } shared_ptr RemoteAtomDBPeer::get_cached_atom(const string& handle) { - return cache_.get_atom(handle); + return cache()->get_atom(handle); } shared_ptr RemoteAtomDBPeer::get_cached_node(const string& handle) { - return cache_.get_node(handle); + return cache()->get_node(handle); } shared_ptr RemoteAtomDBPeer::get_cached_link(const string& handle) { - return cache_.get_link(handle); + return cache()->get_link(handle); } vector> RemoteAtomDBPeer::get_matching_atoms(bool is_toplevel, Atom& key) { @@ -150,7 +114,7 @@ vector> RemoteAtomDBPeer::get_matching_atoms(bool is_toplevel, } }; - merge_results(cache_.get_matching_atoms(is_toplevel, key)); + merge_results(cache()->get_matching_atoms(is_toplevel, key)); if (local_persistence_) { merge_results(local_persistence_->get_matching_atoms(is_toplevel, key)); } @@ -162,27 +126,20 @@ vector> RemoteAtomDBPeer::get_matching_atoms(bool is_toplevel, return result; } -bool RemoteAtomDBPeer::schema_already_fetched(const LinkSchema& link_schema) { - return fetched_link_templates_.lookup(link_schema.handle()) != NULL; -} - void RemoteAtomDBPeer::feed_cache_from_handle_set(shared_ptr handle_set) { if (!handle_set) return; auto it = handle_set->get_iterator(); if (!it) return; - LOG_DEBUG("[" << uid_ << "] feed_cache_from_handle_set: warming cache with " << handle_set->size() - << " handles"); + LOG_DEBUG("[RemoteDB(" << uid_ << ")] feed_cache_from_handle_set: warming cache with " + << handle_set->size() << " handles"); while (true) { char* handle_cstr = it->next(); if (!handle_cstr) break; string handle(handle_cstr); - auto atom = get_atom(handle); - if (atom) { - cache_.add_atom(atom.get()); - } + get_atom(handle); } } @@ -212,88 +169,114 @@ shared_ptr RemoteAtomDBPeer::query_for_pattern(const LinkSchema& link auto result = make_shared(); set seen; - if (schema_already_fetched(link_schema)) { - LOG_DEBUG("[" << uid_ << "] query_for_pattern(" << link_schema.handle() << ") cache-hit" - << (local_persistence_ ? ", merging cache + local_persistence" : ".")); + auto merge_cache = [&]() { + auto snapshot = cache(); merge_handle_set( - cache_.query_for_pattern(link_schema), result, seen, cache_.allow_nested_indexing()); - if (local_persistence_) { - merge_handle_set(local_persistence_->query_for_pattern(link_schema), - result, - seen, - local_persistence_->allow_nested_indexing()); - } - } else { - LOG_DEBUG("[" << uid_ << "] query_for_pattern(" << link_schema.handle() - << ") cache-miss, fetching from remote atomdb"); - if (atomdb_) { - auto handle_set = atomdb_->query_for_pattern(link_schema); - if (handle_set) { - merge_handle_set(handle_set, result, seen, atomdb_->allow_nested_indexing()); - } + snapshot->query_for_pattern(link_schema), result, seen, snapshot->allow_nested_indexing()); + }; + + auto merge_local_persistence = [&]() { + if (!local_persistence_) return; + merge_handle_set(local_persistence_->query_for_pattern(link_schema), + result, + seen, + local_persistence_->allow_nested_indexing()); + }; + + bool cache_hit; + { + lock_guard lock(peer_mutex_); + cache_hit = fetched_link_templates_->lookup(link_schema.handle()) != nullptr; + } + + if (cache_hit) { + LOG_DEBUG("[RemoteDB(" << uid_ << ")] query_for_pattern(" << link_schema.handle() + << ") cache-hit"); + merge_cache(); + merge_local_persistence(); + LOG_DEBUG("[RemoteDB(" << uid_ << ")] query_for_pattern(" << link_schema.handle() << ") -> " + << result->size() << " handles"); + return result; + } + + LOG_DEBUG("[RemoteDB(" << uid_ << ")] query_for_pattern(" << link_schema.handle() + << ") cache-miss, fetching from remote atomdb"); + + shared_ptr remote_handle_set = atomdb_->query_for_pattern(link_schema); + if (remote_handle_set) { + feed_cache_from_handle_set(remote_handle_set); + // Merge remote first when it carries nested metadata. Cache (InMemoryDB) has no + // metta/assignments — if it fills `seen` first, the remote metadata is skipped. + merge_handle_set(remote_handle_set, result, seen, atomdb_->allow_nested_indexing()); + } + + merge_cache(); + merge_local_persistence(); + + { + lock_guard lock(peer_mutex_); + if (fetched_link_templates_->lookup(link_schema.handle()) == nullptr) { + // HandleTrie::insert takes ownership and may delete on re-insert — never reuse one + // EmptyTrieValue* across inserts. + fetched_link_templates_->insert(link_schema.handle(), new EmptyTrieValue()); } - feed_cache_from_handle_set(result); - fetched_link_templates_.insert(link_schema.handle(), empty_trie_value_); } - LOG_DEBUG("[" << uid_ << "] query_for_pattern(" << link_schema.handle() << ") -> " << result->size() - << " handles"); + LOG_DEBUG("[RemoteDB(" << uid_ << ")] query_for_pattern(" << link_schema.handle() << ") -> " + << result->size() << " handles"); return result; } shared_ptr RemoteAtomDBPeer::query_for_targets(const string& handle) { - auto result = cache_.query_for_targets(handle); - if (result) return result; + if (auto result = cache()->query_for_targets(handle)) { + return result; + } if (local_persistence_) { - result = local_persistence_->query_for_targets(handle); + auto result = local_persistence_->query_for_targets(handle); if (result) { - LOG_DEBUG("[" << uid_ << "] query_for_targets(" << handle << ") <- local_persistence"); + LOG_DEBUG("[RemoteDB(" << uid_ << ")] query_for_targets(" << handle + << ") <- local_persistence"); return result; } } - if (atomdb_) { - LOG_DEBUG("[" << uid_ << "] query_for_targets(" << handle << ") <- remote atomdb"); - return atomdb_->query_for_targets(handle); - } - - return nullptr; + LOG_DEBUG("[RemoteDB(" << uid_ << ")] query_for_targets(" << handle << ") <- remote atomdb"); + return atomdb_->query_for_targets(handle); } shared_ptr RemoteAtomDBPeer::query_for_incoming_set(const string& handle) { auto result = make_shared(); set seen; - merge_handle_set(cache_.query_for_incoming_set(handle), result, seen); + merge_handle_set(cache()->query_for_incoming_set(handle), result, seen); if (local_persistence_) { merge_handle_set(local_persistence_->query_for_incoming_set(handle), result, seen); } - if (atomdb_) { - merge_handle_set(atomdb_->query_for_incoming_set(handle), result, seen); - } - LOG_DEBUG("[" << uid_ << "] query_for_incoming_set(" << handle << ") -> " << result->size() - << " handles"); + merge_handle_set(atomdb_->query_for_incoming_set(handle), result, seen); + + LOG_DEBUG("[RemoteDB(" << uid_ << ")] query_for_incoming_set(" << handle << ") -> " << result->size() + << " handles"); return result; } bool RemoteAtomDBPeer::atom_exists(const string& handle) { - if (cache_.atom_exists(handle)) return true; + if (cache()->atom_exists(handle)) return true; if (local_persistence_ && local_persistence_->atom_exists(handle)) return true; if (atomdb_ && atomdb_->atom_exists(handle)) return true; return false; } bool RemoteAtomDBPeer::node_exists(const string& handle) { - if (cache_.node_exists(handle)) return true; + if (cache()->node_exists(handle)) return true; if (local_persistence_ && local_persistence_->node_exists(handle)) return true; if (atomdb_ && atomdb_->node_exists(handle)) return true; return false; } bool RemoteAtomDBPeer::link_exists(const string& handle) { - if (cache_.link_exists(handle)) return true; + if (cache()->link_exists(handle)) return true; if (local_persistence_ && local_persistence_->link_exists(handle)) return true; if (atomdb_ && atomdb_->link_exists(handle)) return true; return false; @@ -301,25 +284,20 @@ bool RemoteAtomDBPeer::link_exists(const string& handle) { set RemoteAtomDBPeer::atoms_exist(const vector& handles) { set result; - set remaining(handles.begin(), handles.end()); + auto from_source = [&](AtomDB& db) { vector to_check(remaining.begin(), remaining.end()); if (to_check.empty()) return; - auto found = db.atoms_exist(to_check); - for (const auto& h : found) { + for (const auto& h : db.atoms_exist(to_check)) { result.insert(h); remaining.erase(h); } }; - from_source(cache_); - if (local_persistence_ && !remaining.empty()) { - from_source(*local_persistence_); - } - if (atomdb_ && !remaining.empty()) { - from_source(*atomdb_); - } + from_source(*cache()); + if (local_persistence_ && !remaining.empty()) from_source(*local_persistence_); + if (atomdb_ && !remaining.empty()) from_source(*atomdb_); return result; } @@ -331,14 +309,13 @@ set RemoteAtomDBPeer::nodes_exist(const vector& handles) { auto from_source = [&](AtomDB& db) { vector to_check(remaining.begin(), remaining.end()); if (to_check.empty()) return; - auto found = db.nodes_exist(to_check); - for (const auto& h : found) { + for (const auto& h : db.nodes_exist(to_check)) { result.insert(h); remaining.erase(h); } }; - from_source(cache_); + from_source(*cache()); if (local_persistence_ && !remaining.empty()) from_source(*local_persistence_); if (atomdb_ && !remaining.empty()) from_source(*atomdb_); @@ -352,14 +329,13 @@ set RemoteAtomDBPeer::links_exist(const vector& handles) { auto from_source = [&](AtomDB& db) { vector to_check(remaining.begin(), remaining.end()); if (to_check.empty()) return; - auto found = db.links_exist(to_check); - for (const auto& h : found) { + for (const auto& h : db.links_exist(to_check)) { result.insert(h); remaining.erase(h); } }; - from_source(cache_); + from_source(*cache()); if (local_persistence_ && !remaining.empty()) from_source(*local_persistence_); if (atomdb_ && !remaining.empty()) from_source(*atomdb_); @@ -367,99 +343,182 @@ set RemoteAtomDBPeer::links_exist(const vector& handles) { } string RemoteAtomDBPeer::add_atom(const atoms::Atom* atom, bool throw_if_exists) { - return cache_.add_atom(atom, throw_if_exists); + vector atoms = {const_cast(atom)}; + auto handles = add_atoms(atoms, throw_if_exists); + return handles.empty() ? "" : handles[0]; } string RemoteAtomDBPeer::add_node(const atoms::Node* node, bool throw_if_exists) { - return cache_.add_node(node, throw_if_exists); + vector nodes = {const_cast(node)}; + auto handles = add_nodes(nodes, throw_if_exists); + return handles.empty() ? "" : handles[0]; } string RemoteAtomDBPeer::add_link(const atoms::Link* link, bool throw_if_exists) { - return cache_.add_link(link, throw_if_exists); + vector links = {const_cast(link)}; + auto handles = add_links(links, throw_if_exists); + return handles.empty() ? "" : handles[0]; } +// Writes hold peer_mutex_ for the whole operation so the cache write, the staged_handles_ +// update and any concurrent release_cache() swap stay mutually consistent (a write must land +// in the same cache generation as its staged handle). Cache ops are pure in-memory and fast. + vector RemoteAtomDBPeer::add_atoms(const vector& atoms, bool throw_if_exists, bool is_transactional) { - return cache_.add_atoms(atoms, throw_if_exists, is_transactional); + if (is_readonly()) return vector(); + + lock_guard lock(peer_mutex_); + fetched_link_templates_ = make_unique(HANDLE_HASH_SIZE - 1); + auto handles = cache_->add_atoms(atoms, throw_if_exists, is_transactional); + staged_handles_.insert(handles.begin(), handles.end()); + return handles; } vector RemoteAtomDBPeer::add_nodes(const vector& nodes, bool throw_if_exists, bool is_transactional) { - return cache_.add_nodes(nodes, throw_if_exists, is_transactional); + if (is_readonly()) return vector(); + + lock_guard lock(peer_mutex_); + fetched_link_templates_ = make_unique(HANDLE_HASH_SIZE - 1); + auto handles = cache_->add_nodes(nodes, throw_if_exists, is_transactional); + staged_handles_.insert(handles.begin(), handles.end()); + return handles; } vector RemoteAtomDBPeer::add_links(const vector& links, bool throw_if_exists, bool is_transactional) { - return cache_.add_links(links, throw_if_exists, is_transactional); + if (is_readonly()) return vector(); + + lock_guard lock(peer_mutex_); + fetched_link_templates_ = make_unique(HANDLE_HASH_SIZE - 1); + auto handles = cache_->add_links(links, throw_if_exists, is_transactional); + staged_handles_.insert(handles.begin(), handles.end()); + return handles; } bool RemoteAtomDBPeer::delete_atom(const string& handle, bool delete_link_targets) { - bool cache_ok = cache_.delete_atom(handle, delete_link_targets); - bool local_ok = true; - if (local_persistence_) { - local_ok = local_persistence_->delete_atom(handle, delete_link_targets); + if (is_readonly()) return false; + + bool cache_ok; + { + lock_guard lock(peer_mutex_); + cache_ok = cache_->delete_atom(handle, delete_link_targets); + if (cache_ok) staged_handles_.erase(handle); + } + bool local_ok = local_persistence_->delete_atom(handle, delete_link_targets); + if (cache_ok || local_ok) { + lock_guard lock(peer_mutex_); + fetched_link_templates_ = make_unique(HANDLE_HASH_SIZE - 1); } return cache_ok || local_ok; } bool RemoteAtomDBPeer::delete_node(const string& handle, bool delete_link_targets) { - bool cache_ok = cache_.delete_node(handle, delete_link_targets); - bool local_ok = true; - if (local_persistence_) { - local_ok = local_persistence_->delete_node(handle, delete_link_targets); + if (is_readonly()) return false; + + bool cache_ok; + { + lock_guard lock(peer_mutex_); + cache_ok = cache_->delete_node(handle, delete_link_targets); + if (cache_ok) staged_handles_.erase(handle); + } + bool local_ok = local_persistence_->delete_node(handle, delete_link_targets); + if (cache_ok || local_ok) { + lock_guard lock(peer_mutex_); + fetched_link_templates_ = make_unique(HANDLE_HASH_SIZE - 1); } return cache_ok || local_ok; } bool RemoteAtomDBPeer::delete_link(const string& handle, bool delete_link_targets) { - bool cache_ok = cache_.delete_link(handle, delete_link_targets); - bool local_ok = true; - if (local_persistence_) { - local_ok = local_persistence_->delete_link(handle, delete_link_targets); + if (is_readonly()) return false; + + bool cache_ok; + { + lock_guard lock(peer_mutex_); + cache_ok = cache_->delete_link(handle, delete_link_targets); + if (cache_ok) staged_handles_.erase(handle); + } + bool local_ok = local_persistence_->delete_link(handle, delete_link_targets); + if (cache_ok || local_ok) { + lock_guard lock(peer_mutex_); + fetched_link_templates_ = make_unique(HANDLE_HASH_SIZE - 1); } return cache_ok || local_ok; } uint RemoteAtomDBPeer::delete_atoms(const vector& handles, bool delete_link_targets) { - uint cache_count = cache_.delete_atoms(handles, delete_link_targets); - uint local_count = 0; - if (local_persistence_) { - local_count = local_persistence_->delete_atoms(handles, delete_link_targets); + if (is_readonly()) return 0; + + uint cache_count; + { + lock_guard lock(peer_mutex_); + cache_count = cache_->delete_atoms(handles, delete_link_targets); + if (cache_count > 0) { + for (const auto& h : handles) staged_handles_.erase(h); + } + } + uint local_count = local_persistence_->delete_atoms(handles, delete_link_targets); + if (cache_count > 0 || local_count > 0) { + lock_guard lock(peer_mutex_); + fetched_link_templates_ = make_unique(HANDLE_HASH_SIZE - 1); } return cache_count + local_count; } uint RemoteAtomDBPeer::delete_nodes(const vector& handles, bool delete_link_targets) { - uint cache_count = cache_.delete_nodes(handles, delete_link_targets); - uint local_count = 0; - if (local_persistence_) { - local_count = local_persistence_->delete_nodes(handles, delete_link_targets); + if (is_readonly()) return 0; + + uint cache_count; + { + lock_guard lock(peer_mutex_); + cache_count = cache_->delete_nodes(handles, delete_link_targets); + if (cache_count > 0) { + for (const auto& h : handles) staged_handles_.erase(h); + } + } + uint local_count = local_persistence_->delete_nodes(handles, delete_link_targets); + if (cache_count > 0 || local_count > 0) { + lock_guard lock(peer_mutex_); + fetched_link_templates_ = make_unique(HANDLE_HASH_SIZE - 1); } return cache_count + local_count; } uint RemoteAtomDBPeer::delete_links(const vector& handles, bool delete_link_targets) { - uint cache_count = cache_.delete_links(handles, delete_link_targets); - uint local_count = 0; - if (local_persistence_) { - local_count = local_persistence_->delete_links(handles, delete_link_targets); + if (is_readonly()) return 0; + + uint cache_count; + { + lock_guard lock(peer_mutex_); + cache_count = cache_->delete_links(handles, delete_link_targets); + if (cache_count > 0) { + for (const auto& h : handles) staged_handles_.erase(h); + } + } + uint local_count = local_persistence_->delete_links(handles, delete_link_targets); + if (cache_count > 0 || local_count > 0) { + lock_guard lock(peer_mutex_); + fetched_link_templates_ = make_unique(HANDLE_HASH_SIZE - 1); } return cache_count + local_count; } void RemoteAtomDBPeer::re_index_patterns(bool flush_patterns) { - cache_.re_index_patterns(flush_patterns); + if (is_readonly()) return; + + cache()->re_index_patterns(flush_patterns); if (local_persistence_) { local_persistence_->re_index_patterns(flush_patterns); } } size_t RemoteAtomDBPeer::node_count() const { - size_t count = 0; - count += cache_.node_count(); + size_t count = cache()->node_count(); if (local_persistence_) { count += local_persistence_->node_count(); } @@ -467,8 +526,7 @@ size_t RemoteAtomDBPeer::node_count() const { } size_t RemoteAtomDBPeer::link_count() const { - size_t count = 0; - count += cache_.link_count(); + size_t count = cache()->link_count(); if (local_persistence_) { count += local_persistence_->link_count(); } @@ -476,8 +534,7 @@ size_t RemoteAtomDBPeer::link_count() const { } size_t RemoteAtomDBPeer::atom_count() const { - size_t count = 0; - count += cache_.atom_count(); + size_t count = cache()->atom_count(); if (local_persistence_) { count += local_persistence_->atom_count(); } @@ -485,37 +542,233 @@ size_t RemoteAtomDBPeer::atom_count() const { } void RemoteAtomDBPeer::fetch(const LinkSchema& link_schema) { - if (!schema_already_fetched(link_schema) && atomdb_) { - LOG_DEBUG("[" << uid_ << "] fetch(" << link_schema.handle() - << ") prefetching from remote atomdb"); - auto result = atomdb_->query_for_pattern(link_schema); - if (result) { - feed_cache_from_handle_set(result); - fetched_link_templates_.insert(link_schema.handle(), empty_trie_value_); + { + lock_guard lock(peer_mutex_); + if (fetched_link_templates_->lookup(link_schema.handle()) != nullptr) { + return; } } + + LOG_DEBUG("[RemoteDB(" << uid_ << ")] fetch(" << link_schema.handle() + << ") prefetching from remote atomdb"); + auto result = atomdb_->query_for_pattern(link_schema); + if (!result) { + return; + } + + feed_cache_from_handle_set(result); + + lock_guard lock(peer_mutex_); + if (fetched_link_templates_->lookup(link_schema.handle()) == nullptr) { + fetched_link_templates_->insert(link_schema.handle(), new EmptyTrieValue()); + } } -void RemoteAtomDBPeer::release(const LinkSchema& link_schema) { - // Query cache for pattern results before removing, so we can persist them. - // TODO: This may remove atoms that were also fetched by another schema. We need to handle this - // properly. - LOG_DEBUG("[" << uid_ << "] release(" << link_schema.handle() - << ") evicting from cache to local_persistence"); - auto handle_set = cache_.query_for_pattern(link_schema); - if (handle_set && local_persistence_) { - auto it = handle_set->get_iterator(); - char* handle_cstr; - while ((handle_cstr = it->next()) != nullptr) { - string handle(handle_cstr); - auto atom = cache_.get_atom(handle); - if (atom) { - local_persistence_->add_atom(atom.get(), false); - cache_.delete_atom(handle, false); +void RemoteAtomDBPeer::persist_atoms_to_local(const vector>& atoms) { + if (!local_persistence_ || atoms.empty()) { + return; + } + + // Build dependency closure from this peer's own backends only (cache already flushed into + // `atoms`, plus atomdb_ / local_persistence). Cross-peer federation is intentionally not + // attempted here; local_persistence is expected to run with composite_type_enabled=false so + // links can be written without resolving foreign targets for composite_type_hash. + unordered_map> atoms_by_handle; + atoms_by_handle.reserve(atoms.size()); + for (const auto& atom : atoms) { + if (atom) { + atoms_by_handle[atom->handle()] = atom; + } + } + + deque missing_targets_queue; + auto enqueue_missing_targets = [&](const shared_ptr& atom) { + if (!atom || atom->arity() == 0) return; + auto* link = dynamic_cast(atom.get()); + if (link == nullptr) return; + for (const auto& target : link->targets) { + if (atoms_by_handle.find(target) == atoms_by_handle.end() && + !local_persistence_->atom_exists(target)) { + missing_targets_queue.push_back(target); } } + }; + + for (const auto& [_, atom] : atoms_by_handle) { + enqueue_missing_targets(atom); + } + + while (!missing_targets_queue.empty()) { + string target_handle = missing_targets_queue.front(); + missing_targets_queue.pop_front(); + + if (atoms_by_handle.find(target_handle) != atoms_by_handle.end() || + local_persistence_->atom_exists(target_handle)) { + continue; + } + + auto fetched_atom = atomdb_->get_atom(target_handle); + if (!fetched_atom) { + LOG_DEBUG("[RemoteDB(" << uid_ << ")] persist_atoms_to_local: target " << target_handle + << " not available on this peer (composite_type disabled peers " + "may still persist the link)"); + continue; + } + atoms_by_handle[target_handle] = fetched_atom; + enqueue_missing_targets(fetched_atom); + } + + vector nodes; + vector links; + nodes.reserve(atoms_by_handle.size()); + links.reserve(atoms_by_handle.size()); + for (const auto& [_, atom] : atoms_by_handle) { + if (!atom) continue; + if (atom->arity() == 0) { + auto* node = dynamic_cast(atom.get()); + if (node != nullptr) nodes.push_back(node); + } else { + auto* link = dynamic_cast(atom.get()); + if (link != nullptr) links.push_back(link); + } + } + + // Overwrite in place via the backend's upsert semantics (RedisMongoDB replace_one upsert=true). + // Do NOT delete first: RedisMongoDB::delete_atom cascades over the atom's incoming set, which + // would wipe large parts of the base KB when a staged handle is a shared node. + if (!nodes.empty()) { + local_persistence_->add_nodes(nodes, false, false); + } + + if (links.empty()) { + return; + } + + // When composite_type is disabled, backends skip target-existence checks — persist links + // directly even if some targets live only on other peers. + if (!local_persistence_->composite_type_enabled()) { + local_persistence_->add_links(links, false, false); + return; + } + + // Persist links in dependency-safe batches when composite_type requires local targets. + vector remaining_links = links; + while (!remaining_links.empty()) { + vector batch; + vector pending; + for (auto* link : remaining_links) { + bool all_targets_exist = true; + for (const auto& target : link->targets) { + if (!local_persistence_->atom_exists(target)) { + all_targets_exist = false; + break; + } + } + if (all_targets_exist) { + batch.push_back(link); + } else { + pending.push_back(link); + } + } + if (batch.empty()) { + LOG_ERROR("[RemoteDB(" << uid_ << ")] persist_atoms_to_local: dropping " << pending.size() + << " links with unresolved targets (writes will be lost after " + "cache release)"); + break; + } + local_persistence_->add_links(batch, false, false); + remaining_links = pending; + } +} + +void RemoteAtomDBPeer::release_cache(bool persist_to_local, bool persist_entire_cache) { + vector> to_persist; + + { + lock_guard lock(peer_mutex_); + if (cache_->atom_count() == 0 && fetched_link_templates_->size == 0) { + return; + } + + LOG_DEBUG("[RemoteDB(" << uid_ << ")] release_cache: clearing " << cache_->atom_count() + << " cached atoms" + << (persist_entire_cache ? " (persist entire cache)" + : " (persist staged only)")); + + if (local_persistence_ && cache_->atom_count() > 0) { + // Invariant: staged writes are never dropped. Even a "no persistence" release + // flushes them, so a write staged between release()'s decision and this critical + // section cannot be lost. + if (persist_to_local && persist_entire_cache) { + to_persist = cache_->get_all_atoms(); + } else if (!staged_handles_.empty()) { + for (const auto& atom : cache_->get_all_atoms()) { + if (staged_handles_.count(atom->handle()) > 0) { + to_persist.push_back(atom); + } + } + } + } + + // Swap in a fresh cache; concurrent readers holding the old snapshot keep a live, + // internally thread-safe InMemoryDB until they drop it. + cache_ = make_shared(uid_); + staged_handles_.clear(); + fetched_link_templates_ = make_unique(HANDLE_HASH_SIZE - 1); + } + + if (!to_persist.empty()) { + persist_atoms_to_local(to_persist); + } +} + +void RemoteAtomDBPeer::release(const LinkSchema& link_schema, bool persist, bool force) { + bool schema_cached; + bool has_staged_writes; + bool has_cached_atoms; + { + lock_guard lock(peer_mutex_); + schema_cached = fetched_link_templates_->lookup(link_schema.handle()) != nullptr; + has_staged_writes = !staged_handles_.empty(); + has_cached_atoms = cache_->atom_count() > 0; + } + + // This snapshot is advisory: state may change before release_cache() re-locks. That is safe + // because release_cache() never drops staged writes (see invariant there). + if (persist) { + if (schema_cached || force) { + LOG_INFO("[RemoteDB(" << uid_ << ")] release(" << link_schema.handle() + << ") clearing cache with persistence (or forced)"); + release_cache(true, true); + return; + } + if (has_staged_writes) { + LOG_INFO("[RemoteDB(" << uid_ << ")] release(" << link_schema.handle() + << ") persisting staged cache writes"); + release_cache(true, false); + return; + } + if (!is_readonly() && has_cached_atoms) { + // Important for memory pressure: callers may request a flush even when this exact + // schema token is not marked as fetched anymore. + LOG_INFO("[RemoteDB(" << uid_ << ")] release(" << link_schema.handle() + << ") clearing read cache without persistence"); + release_cache(false, true); + return; + } + LOG_DEBUG( + "[RemoteDB(" << uid_ << ")] release(" << link_schema.handle() + << ") no-op (schema not cached, no staged writes, empty cache or readonly)"); + return; + } + + if (schema_cached) { + lock_guard lock(peer_mutex_); + fetched_link_templates_->remove(link_schema.handle()); + } else { + LOG_DEBUG("[RemoteDB(" << uid_ << ")] release(" << link_schema.handle() + << ") cache already released"); } - fetched_link_templates_.remove(link_schema.handle(), false); } double RemoteAtomDBPeer::available_ram() { @@ -524,12 +777,12 @@ double RemoteAtomDBPeer::available_ram() { return static_cast(Utils::get_current_free_ram()) / static_cast(total); } -// TODO: Implement actual cache eviction void RemoteAtomDBPeer::auto_cleanup() { double ram_fraction = available_ram(); if (ram_fraction < CRITICAL_RAM_THRESHOLD) { LOG_INFO("RemoteAtomDBPeer " << uid_ << ": Low RAM detected (available=" - << (ram_fraction * 100.0) << "%), cache cleanup triggered"); + << (ram_fraction * 100.0) << "%), cache release triggered"); + release_cache(true, true); } } diff --git a/src/atomdb/remotedb/RemoteAtomDBPeer.h b/src/atomdb/remotedb/RemoteAtomDBPeer.h index 0c5916fe..b457e1e2 100644 --- a/src/atomdb/remotedb/RemoteAtomDBPeer.h +++ b/src/atomdb/remotedb/RemoteAtomDBPeer.h @@ -2,6 +2,7 @@ #include #include +#include #include #include "AtomDB.h" @@ -17,14 +18,19 @@ using namespace atoms; namespace atomdb { /** - * RemoteAtomDBPeer represents a cached connection to a remote AtomDB. - * It combines an in-memory cache, a read-only remote AtomDB, and local persistence - * for newly added atoms. + * RemoteAtomDBPeer represents a connection to a remote AtomDB with optional layers. + * It may combine an in-memory cache, a remote AtomDB backend, and optional local persistence + * for newly added atoms. Peers without local_persistence are readonly for mutations; get_atom + * prefers local_persistence over cache so updated custom attributes (e.g. strength) are visible. + * + * Thread-safety: the cache (InMemoryDB) is internally thread-safe, so cache reads/writes need + * no locking here. peer_mutex_ only guards the peer's own bookkeeping: the cache_ pointer swap + * (release_cache), fetched_link_templates_ and staged_handles_. */ class RemoteAtomDBPeer : public AtomDB, public processor::ThreadMethod { public: RemoteAtomDBPeer(shared_ptr remote_atomdb, - shared_ptr local_persistence, + shared_ptr local_persistence = nullptr, const string& uid = ""); ~RemoteAtomDBPeer(); @@ -86,7 +92,8 @@ class RemoteAtomDBPeer : public AtomDB, public processor::ThreadMethod { // Cache policy API void fetch(const LinkSchema& link_schema); - void release(const LinkSchema& link_schema); + void release(const LinkSchema& link_schema, bool persist = true, bool force = false); + void release_cache(bool persist_to_local, bool persist_entire_cache); double available_ram(); void auto_cleanup(); void start_cleanup_thread(); @@ -96,26 +103,34 @@ class RemoteAtomDBPeer : public AtomDB, public processor::ThreadMethod { bool thread_one_step() override; const string& get_uid() const { return uid_; } - bool is_readonly() const { return local_persistence_ == nullptr; } private: + // Snapshot of cache_. Safe to use outside peer_mutex_ because InMemoryDB is internally + // thread-safe; release_cache() swaps the pointer under the mutex, and old snapshots stay + // alive (shared_ptr) until their holders finish. + shared_ptr cache() const; + void feed_cache_from_handle_set(shared_ptr handle_set); void merge_handle_set(shared_ptr source, shared_ptr dest, set& seen, bool copy_metadata = false); - bool schema_already_fetched(const LinkSchema& link_schema); + void persist_atoms_to_local(const vector>& atoms); string uid_; - InMemoryDB cache_; + shared_ptr cache_; shared_ptr atomdb_; shared_ptr local_persistence_; - HandleTrie fetched_link_templates_; - EmptyTrieValue* empty_trie_value_; - + unique_ptr fetched_link_templates_; + set staged_handles_; unique_ptr cleanup_thread_; + // Guards only the peer's bookkeeping: cache_ pointer swap, fetched_link_templates_ and + // staged_handles_. Atom data itself is protected inside InMemoryDB. Never held across + // atomdb_ / local_persistence_ I/O. + mutable mutex peer_mutex_; + static constexpr double CRITICAL_RAM_THRESHOLD = 0.1; // 10% - cleanup when below this }; diff --git a/src/tests/assets/remotedb_config_single.json b/src/tests/assets/remotedb_config_single.json index c92e6710..eef87934 100644 --- a/src/tests/assets/remotedb_config_single.json +++ b/src/tests/assets/remotedb_config_single.json @@ -13,6 +13,10 @@ "endpoint": "localhost:40021", "username": "admin", "password": "admin" + }, + "local_persistence": { + "type": "inmemorydb", + "context": "remotedb_test_single_peer_local_" } } ] diff --git a/src/tests/cpp/BUILD b/src/tests/cpp/BUILD index 8d7a3ae7..bca30860 100644 --- a/src/tests/cpp/BUILD +++ b/src/tests/cpp/BUILD @@ -889,7 +889,7 @@ cc_test( cc_test( name = "remote_atomdb_test", - size = "small", + size = "medium", srcs = ["remote_atomdb_test.cc"], copts = [ "-Iexternal/gtest/googletest/include", diff --git a/src/tests/cpp/inmemorydb_test.cc b/src/tests/cpp/inmemorydb_test.cc index 234d0651..a1bd74a0 100644 --- a/src/tests/cpp/inmemorydb_test.cc +++ b/src/tests/cpp/inmemorydb_test.cc @@ -5,6 +5,7 @@ #include #include #include +#include #include #include @@ -13,6 +14,7 @@ #include "Link.h" #include "LinkSchema.h" #include "Node.h" +#include "Properties.h" using namespace atomdb; using namespace atomdb::atomdb_api_types; @@ -541,6 +543,96 @@ TEST_F(InMemoryDBTest, AtomsCount) { EXPECT_EQ(db->empty(), false); } +TEST_F(InMemoryDBTest, GetAllAtoms) { + auto node1 = new Node("Symbol", "Node1"); + auto node2 = new Node("Symbol", "Node2"); + auto similarity = new Node("Symbol", "Similarity"); + + string node1_handle = db->add_node(node1, false); + string node2_handle = db->add_node(node2, false); + string similarity_handle = db->add_node(similarity, false); + + auto link = new Link("Expression", {similarity_handle, node1_handle, node2_handle}); + string link_handle = db->add_link(link, false); + + auto atoms = db->get_all_atoms(); + EXPECT_EQ(atoms.size(), 4u); + + set handles; + for (const auto& atom : atoms) { + handles.insert(atom->handle()); + } + EXPECT_TRUE(handles.count(node1_handle) > 0); + EXPECT_TRUE(handles.count(node2_handle) > 0); + EXPECT_TRUE(handles.count(similarity_handle) > 0); + EXPECT_TRUE(handles.count(link_handle) > 0); +} + +TEST_F(InMemoryDBTest, GetAllAtomsReturnsIndependentClones) { + auto node = new Node("Symbol", "Node1"); + string handle = db->add_node(node, false); + + auto atoms = db->get_all_atoms(); + ASSERT_EQ(atoms.size(), 1u); + EXPECT_EQ(atoms[0]->handle(), handle); + + db->drop_all(); + + // Clones returned earlier remain valid after the DB is cleared. + EXPECT_EQ(atoms[0]->handle(), handle); + EXPECT_EQ(db->get_atom(handle), nullptr); +} + +TEST_F(InMemoryDBTest, DropAllClearsAtomsAndIndexes) { + auto human = new Node("Symbol", "\"human\""); + auto monkey = new Node("Symbol", "\"monkey\""); + auto mammal = new Node("Symbol", "\"mammal\""); + auto inheritance = new Node("Symbol", "Inheritance"); + + string human_handle = db->add_node(human, false); + string monkey_handle = db->add_node(monkey, false); + string mammal_handle = db->add_node(mammal, false); + string inheritance_handle = db->add_node(inheritance, false); + + auto link1 = new Link("Expression", {inheritance_handle, human_handle, mammal_handle}); + auto link2 = new Link("Expression", {inheritance_handle, monkey_handle, mammal_handle}); + string link1_handle = db->add_link(link1, false); + string link2_handle = db->add_link(link2, false); + + LinkSchema link_schema({"LINK_TEMPLATE", + "Expression", + "3", + "NODE", + "Symbol", + "Inheritance", + "VARIABLE", + "x", + "NODE", + "Symbol", + "\"mammal\""}); + + EXPECT_EQ(db->query_for_pattern(link_schema)->size(), 2); + EXPECT_EQ(db->query_for_incoming_set(human_handle)->size(), 1); + EXPECT_NE(db->query_for_targets(link1_handle), nullptr); + + db->drop_all(); + + EXPECT_EQ(db->atom_count(), 0u); + EXPECT_EQ(db->empty(), true); + EXPECT_EQ(db->get_atom(human_handle), nullptr); + EXPECT_EQ(db->get_atom(link1_handle), nullptr); + EXPECT_EQ(db->query_for_pattern(link_schema)->size(), 0); + EXPECT_EQ(db->query_for_incoming_set(human_handle)->size(), 0); + EXPECT_EQ(db->query_for_targets(link1_handle), nullptr); +} + +TEST_F(InMemoryDBTest, DropAllOnEmptyDatabase) { + EXPECT_EQ(db->atom_count(), 0u); + EXPECT_EQ(db->get_all_atoms().size(), 0u); + db->drop_all(); + EXPECT_EQ(db->atom_count(), 0u); +} + // ============================================================================= // HandleSetInMemory / HandleSetInMemoryIterator tests // @@ -611,6 +703,38 @@ TEST(HandleSetInMemoryTest, AppendPreservesMetadataFromBothSets) { EXPECT_EQ(first->get_assignments_by_handle(handle_b).get("$b"), "value_b"); } +TEST_F(InMemoryDBTest, UpsertReplacesCustomAttributes) { + auto a = new Node("Symbol", "\"a\""); + auto b = new Node("Symbol", "\"b\""); + auto similarity = new Node("Symbol", "Similarity"); + string a_h = db->add_node(a, false); + string b_h = db->add_node(b, false); + string sim_h = db->add_node(similarity, false); + + auto weak = new Link("Expression", {sim_h, a_h, b_h}, true, Properties{{"strength", 0.5}}); + string handle = db->add_link(weak, false); + auto got_weak = db->get_atom(handle); + ASSERT_NE(got_weak, nullptr); + EXPECT_DOUBLE_EQ(got_weak->custom_attributes.get_or("strength", -1.0), 0.5); + + auto strong = new Link("Expression", {sim_h, a_h, b_h}, true, Properties{{"strength", 0.9}}); + EXPECT_EQ(db->add_link(strong, false), handle); + auto got_strong = db->get_atom(handle); + ASSERT_NE(got_strong, nullptr); + EXPECT_DOUBLE_EQ(got_strong->custom_attributes.get_or("strength", -1.0), 0.9); + + // Incoming set must not grow from the upsert (same content-addressed handle/targets). + auto incoming = db->query_for_incoming_set(a_h); + ASSERT_NE(incoming, nullptr); + EXPECT_EQ(incoming->size(), 1u); + + delete weak; + delete strong; + delete a; + delete b; + delete similarity; +} + TEST(HandleSetInMemoryTest, IteratorPointerOutlivesIterator) { auto set = make_shared(); const string handle = "0123456789abcdef0123456789abcdef"; diff --git a/src/tests/cpp/redis_mongodb_test.cc b/src/tests/cpp/redis_mongodb_test.cc index ff0e7c4f..c0375920 100644 --- a/src/tests/cpp/redis_mongodb_test.cc +++ b/src/tests/cpp/redis_mongodb_test.cc @@ -4,6 +4,8 @@ #include #include #include +#include +#include #include #include @@ -993,6 +995,43 @@ TEST_F(RedisMongoDBTest, AddLinksWithDuplicateTargets) { EXPECT_EQ(db->delete_link(link->handle(), true), true); } +TEST_F(RedisMongoDBTest, CompositeHashDisabledSkipsTargetChecks) { + // Default (composite_type_enabled=true) rejects links whose targets are missing locally. + auto missing_a = new Node("Symbol", "MissingCompositeTargetA"); + auto missing_b = new Node("Symbol", "MissingCompositeTargetB"); + auto implication = new Node("Symbol", "ImplicationMissingComposite"); + auto link = new Link("Expression", + {implication->handle(), missing_a->handle(), missing_b->handle()}, + true, + Properties{{"strength", 0.833333}}); + EXPECT_THROW(db->add_links({link}, false, false), runtime_error); + + // Same Redis/Mongo namespace, composite_type_enabled=false: skip checks and persist the link. + auto json = nlohmann::json(); + json["type"] = "redismongodb"; + json["composite_type_enabled"] = false; + json["redis"] = {{"endpoint", "localhost:40020"}, {"cluster", false}}; + json["mongodb"] = {{"endpoint", "localhost:40021"}, {"username", "admin"}, {"password", "admin"}}; + auto local_db = make_shared("test_", false, commons::JsonConfig(json)); + + auto handles = local_db->add_links({link}, false, false); + ASSERT_EQ(handles.size(), 1); + EXPECT_EQ(handles[0], link->handle()); + ASSERT_TRUE(local_db->link_exists(link->handle())); + EXPECT_FALSE(local_db->atom_exists(missing_a->handle())); + EXPECT_FALSE(local_db->atom_exists(missing_b->handle())); + + auto got = local_db->get_atom(link->handle()); + ASSERT_NE(got, nullptr); + EXPECT_DOUBLE_EQ(got->custom_attributes.get_or("strength", -1.0), 0.833333); + + EXPECT_TRUE(local_db->delete_link(link->handle(), false)); + delete link; + delete missing_a; + delete missing_b; + delete implication; +} + TEST_F(RedisMongoDBTest, AtomsCount) { db->drop_all(); diff --git a/src/tests/cpp/remote_atomdb_test.cc b/src/tests/cpp/remote_atomdb_test.cc index 76e3f2ce..57221406 100644 --- a/src/tests/cpp/remote_atomdb_test.cc +++ b/src/tests/cpp/remote_atomdb_test.cc @@ -43,6 +43,41 @@ class RemoteAtomDBPeerTest : public ::testing::Test { shared_ptr peer_; }; +// Builds the Inheritance(x, "mammal") pattern used across several cache tests. +static LinkSchema inheritance_mammal_schema() { + return LinkSchema({"LINK_TEMPLATE", + "Expression", + "3", + "NODE", + "Symbol", + "Inheritance", + "VARIABLE", + "x", + "NODE", + "Symbol", + "\"mammal\""}); +} + +// Populates a backend with two Inheritance(*, "mammal") links and returns their handles. +static vector populate_inheritance_mammal_links(shared_ptr backend) { + auto human = new Node("Symbol", "\"human\""); + auto monkey = new Node("Symbol", "\"monkey\""); + auto mammal = new Node("Symbol", "\"mammal\""); + auto inheritance = new Node("Symbol", "Inheritance"); + + string human_handle = backend->add_node(human, false); + string monkey_handle = backend->add_node(monkey, false); + string mammal_handle = backend->add_node(mammal, false); + string inheritance_handle = backend->add_node(inheritance, false); + + auto link1 = new Link("Expression", {inheritance_handle, human_handle, mammal_handle}); + auto link2 = new Link("Expression", {inheritance_handle, monkey_handle, mammal_handle}); + string link1_handle = backend->add_link(link1, false); + string link2_handle = backend->add_link(link2, false); + backend->re_index_patterns(true); + return {link1_handle, link2_handle}; +} + TEST_F(RemoteAtomDBPeerTest, AddAndGetNodes) { auto human = new Node("Symbol", "\"human\""); auto monkey = new Node("Symbol", "\"monkey\""); @@ -98,13 +133,17 @@ TEST_F(RemoteAtomDBPeerTest, GetFromCacheThenRemote) { TEST_F(RemoteAtomDBPeerTest, PersistsToLocal) { auto human = new Node("Symbol", "\"human\""); + auto monkey = new Node("Symbol", "\"monkey\""); string human_handle = peer_->add_node(human, false); + string monkey_handle = peer_->add_node(monkey, false); - // cache should have it - EXPECT_TRUE(peer_->node_exists(human_handle)); - auto from_cache = peer_->get_node(human_handle); - ASSERT_NE(from_cache, nullptr); - EXPECT_EQ(from_cache->handle(), human_handle); + peer_->release_cache(true, true); + + EXPECT_TRUE(local_->node_exists(human_handle)); + EXPECT_EQ(peer_->get_cached_atom(human_handle), nullptr); + + EXPECT_TRUE(local_->node_exists(monkey_handle)); + EXPECT_EQ(peer_->get_cached_atom(monkey_handle), nullptr); } TEST_F(RemoteAtomDBPeerTest, QueryForPattern) { @@ -217,61 +256,110 @@ TEST_F(RemoteAtomDBPeerTest, GetUid) { EXPECT_EQ(peer_->get_uid(), "test_peer"); TEST_F(RemoteAtomDBPeerTest, AllowNestedIndexing) { EXPECT_FALSE(peer_->allow_nested_indexing()); } TEST_F(RemoteAtomDBPeerTest, FetchAndRelease) { - auto human = new Node("Symbol", "\"human\""); - auto monkey = new Node("Symbol", "\"monkey\""); - auto mammal = new Node("Symbol", "\"mammal\""); - auto inheritance = new Node("Symbol", "Inheritance"); - - // Add nodes and links to the remote DB directly (bypass peer) - string human_handle = remote_->add_node(human, false); - string monkey_handle = remote_->add_node(monkey, false); - string mammal_handle = remote_->add_node(mammal, false); - string inheritance_handle = remote_->add_node(inheritance, false); - - auto link1 = new Link("Expression", {inheritance_handle, human_handle, mammal_handle}); - auto link2 = new Link("Expression", {inheritance_handle, monkey_handle, mammal_handle}); - string link1_handle = remote_->add_link(link1, false); - string link2_handle = remote_->add_link(link2, false); - remote_->re_index_patterns(true); + auto handles = populate_inheritance_mammal_links(remote_); + string link1_handle = handles[0]; + string link2_handle = handles[1]; + LinkSchema link_schema = inheritance_mammal_schema(); - LinkSchema link_schema({"LINK_TEMPLATE", - "Expression", - "3", - "NODE", - "Symbol", - "Inheritance", - "VARIABLE", - "x", - "NODE", - "Symbol", - "\"mammal\""}); - - // fetch() should pull the pattern results into the cache peer_->fetch(link_schema); - // The peer should now answer the query from its cache auto result = peer_->query_for_pattern(link_schema); ASSERT_NE(result, nullptr); EXPECT_EQ(result->size(), 2); + EXPECT_NE(peer_->get_cached_atom(link1_handle), nullptr); - // Atoms from the query results should be in the cache (retrievable via peer) - auto atom1 = peer_->get_atom(link1_handle); - ASSERT_NE(atom1, nullptr); - EXPECT_EQ(atom1->handle(), link1_handle); - - // release() should move cached atoms to local_persistence and remove from cache peer_->release(link_schema); - // After release, atoms should be in local_persistence EXPECT_TRUE(local_->atom_exists(link1_handle)); EXPECT_TRUE(local_->atom_exists(link2_handle)); + EXPECT_EQ(peer_->get_cached_atom(link1_handle), nullptr); - // The peer should still find the atoms (via local_persistence fallback) auto after_release = peer_->get_atom(link1_handle); ASSERT_NE(after_release, nullptr); EXPECT_EQ(after_release->handle(), link1_handle); } +TEST_F(RemoteAtomDBPeerTest, AddAfterFetchRefetchesPatternFromRemote) { + auto handles = populate_inheritance_mammal_links(remote_); + string link1_handle = handles[0]; + string link2_handle = handles[1]; + LinkSchema link_schema = inheritance_mammal_schema(); + + peer_->fetch(link_schema); + ASSERT_EQ(peer_->query_for_pattern(link_schema)->size(), 2); + EXPECT_NE(peer_->get_cached_atom(link1_handle), nullptr); + + // Any write invalidates the fetched-template registry but keeps read-cached atoms in cache. + auto staged = new Node("Symbol", "\"staged\""); + string staged_handle = peer_->add_node(staged, false); + EXPECT_NE(peer_->get_cached_atom(link1_handle), nullptr); + EXPECT_NE(peer_->get_cached_atom(link2_handle), nullptr); + EXPECT_NE(peer_->get_cached_atom(staged_handle), nullptr); + + // Remote gains a new matching link after the template registry was invalidated. + auto link1 = peer_->get_link(link1_handle); + ASSERT_NE(link1, nullptr); + string inheritance_handle = link1->targets[0]; + string mammal_handle = link1->targets[2]; + + auto chimp = new Node("Symbol", "\"chimp\""); + string chimp_handle = remote_->add_node(chimp, false); + auto link3 = new Link("Expression", {inheritance_handle, chimp_handle, mammal_handle}); + string link3_handle = remote_->add_link(link3, false); + remote_->re_index_patterns(true); + + auto result = peer_->query_for_pattern(link_schema); + ASSERT_NE(result, nullptr); + EXPECT_EQ(result->size(), 3); + + vector result_handles; + auto it = result->get_iterator(); + char* h; + while ((h = it->next()) != nullptr) { + result_handles.push_back(h); + } + EXPECT_TRUE(find(result_handles.begin(), result_handles.end(), link3_handle) != + result_handles.end()); +} + +TEST_F(RemoteAtomDBPeerTest, DeleteAfterFetchReleasesRemoteCache) { + auto handles = populate_inheritance_mammal_links(remote_); + string link1_handle = handles[0]; + string link2_handle = handles[1]; + LinkSchema link_schema = inheritance_mammal_schema(); + + peer_->fetch(link_schema); + EXPECT_NE(peer_->get_cached_atom(link1_handle), nullptr); + EXPECT_NE(peer_->get_cached_atom(link2_handle), nullptr); + + // Delete removes the target from cache and invalidates the fetched-template registry. + ASSERT_TRUE(peer_->delete_link(link1_handle, false)); + EXPECT_EQ(peer_->get_cached_atom(link1_handle), nullptr); + EXPECT_NE(peer_->get_cached_atom(link2_handle), nullptr); + + auto link2 = peer_->get_link(link2_handle); + ASSERT_NE(link2, nullptr); + + auto chimp = new Node("Symbol", "\"chimp\""); + string chimp_handle = remote_->add_node(chimp, false); + auto link3 = new Link("Expression", {link2->targets[0], chimp_handle, link2->targets[2]}); + string link3_handle = remote_->add_link(link3, false); + remote_->re_index_patterns(true); + + auto result = peer_->query_for_pattern(link_schema); + ASSERT_NE(result, nullptr); + EXPECT_EQ(result->size(), 3); + + vector result_handles; + auto it = result->get_iterator(); + char* h; + while ((h = it->next()) != nullptr) { + result_handles.push_back(h); + } + EXPECT_TRUE(find(result_handles.begin(), result_handles.end(), link3_handle) != + result_handles.end()); +} + TEST_F(RemoteAtomDBPeerTest, ReleaseWithoutLocalPersistence) { // Create a peer without local persistence (nullptr) auto remote = make_shared("no_local_remote_"); @@ -335,6 +423,103 @@ TEST_F(RemoteAtomDBPeerTest, AtomsCount) { EXPECT_EQ(peer_->empty(), false); } +// ============================================================================= +// RemoteAtomDBPeer readonly / persistence tests +// ============================================================================= + +TEST(RemoteAtomDBPeerReadonlyTest, RepeatQuerySeesRemoteAfterRelease) { + auto remote = make_shared("readonly_remote_"); + auto peer = make_shared(remote, nullptr, "readonly_peer"); + + auto handles = populate_inheritance_mammal_links(remote); + string link1_handle = handles[0]; + LinkSchema link_schema = inheritance_mammal_schema(); + + auto result1 = peer->query_for_pattern(link_schema); + ASSERT_NE(result1, nullptr); + EXPECT_EQ(result1->size(), 2u); + + auto link1 = peer->get_link(link1_handle); + ASSERT_NE(link1, nullptr); + string inheritance_handle = link1->targets[0]; + string mammal_handle = link1->targets[2]; + + auto chimp = new Node("Symbol", "\"chimp\""); + string chimp_handle = remote->add_node(chimp, false); + auto link3 = new Link("Expression", {inheritance_handle, chimp_handle, mammal_handle}); + string link3_handle = remote->add_link(link3, false); + remote->re_index_patterns(true); + + // Drop the fetched-schema marker so the next query re-hits the remote backend. + peer->release(link_schema, false); + + auto result2 = peer->query_for_pattern(link_schema); + ASSERT_NE(result2, nullptr); + EXPECT_EQ(result2->size(), 3u); + + vector result_handles; + auto it = result2->get_iterator(); + char* h; + while ((h = it->next()) != nullptr) { + result_handles.push_back(h); + } + EXPECT_TRUE(find(result_handles.begin(), result_handles.end(), link3_handle) != + result_handles.end()); +} + +TEST(RemoteAtomDBPeerReadonlyTest, AddStagesThenPersistsOnRelease) { + auto remote = make_shared("stage_remote_"); + auto local = make_shared("stage_local_"); + auto peer = make_shared(remote, local, "stage_peer"); + + auto human = new Node("Symbol", "\"human\""); + string human_handle = peer->add_node(human, false); + + EXPECT_FALSE(human_handle.empty()); + EXPECT_NE(peer->get_cached_atom(human_handle), nullptr); + EXPECT_FALSE(local->node_exists(human_handle)); + + peer->release_cache(true, false); + EXPECT_TRUE(local->node_exists(human_handle)); + EXPECT_TRUE(peer->node_exists(human_handle)); +} + +TEST(RemoteAtomDBPeerReadonlyTest, AddFailsWithoutLocalPersistence) { + auto remote = make_shared("read_only_remote_"); + auto peer = make_shared(remote, nullptr, "read_only_peer"); + + auto human = new Node("Symbol", "\"human\""); + EXPECT_TRUE(peer->add_node(human, false).empty()); + EXPECT_TRUE(peer->add_link(new Link("Expression", {"a", "b", "c"}), false).empty()); +} + +TEST(RemoteAtomDBPeerReadonlyTest, DeleteAtomFromLocalPersistence) { + auto remote = make_shared("delete_remote_"); + auto local = make_shared("delete_local_"); + auto peer = make_shared(remote, local, "delete_peer"); + + auto human = new Node("Symbol", "\"human\""); + string human_handle = local->add_node(human, false); + + EXPECT_TRUE(peer->delete_atom(human_handle, false)); + EXPECT_FALSE(peer->atom_exists(human_handle)); + EXPECT_FALSE(local->atom_exists(human_handle)); +} + +TEST(RemoteAtomDBPeerReadonlyTest, FetchWarmsCache) { + auto remote = make_shared("fetch_remote_"); + auto peer = make_shared(remote, nullptr, "fetch_peer"); + auto handles = populate_inheritance_mammal_links(remote); + LinkSchema link_schema = inheritance_mammal_schema(); + + peer->fetch(link_schema); + EXPECT_NE(peer->get_cached_atom(handles[0]), nullptr); + + auto result = peer->query_for_pattern(link_schema); + ASSERT_NE(result, nullptr); + EXPECT_EQ(result->size(), 2u); +} + // Resolves path to config file from Bazel runfiles or workspace. string resolve_config_path(const string& filename) { auto exists = [](const string& path) { @@ -539,45 +724,9 @@ class CompositeTypeEnabledInMemoryDB : public InMemoryDB { bool composite_type_enabled() const override { return true; } }; -// Builds an Inheritance(x, "mammal") pattern that matches two links in the -// helper-populated backends below. -static LinkSchema inheritance_mammal_schema() { - return LinkSchema({"LINK_TEMPLATE", - "Expression", - "3", - "NODE", - "Symbol", - "Inheritance", - "VARIABLE", - "x", - "NODE", - "Symbol", - "\"mammal\""}); -} - -// Populates a backend with two Inheritance(*, "mammal") links and returns their handles. -static vector populate_inheritance_links(shared_ptr backend) { - auto human = new Node("Symbol", "\"human\""); - auto monkey = new Node("Symbol", "\"monkey\""); - auto mammal = new Node("Symbol", "\"mammal\""); - auto inheritance = new Node("Symbol", "Inheritance"); - - string human_handle = backend->add_node(human, false); - string monkey_handle = backend->add_node(monkey, false); - string mammal_handle = backend->add_node(mammal, false); - string inheritance_handle = backend->add_node(inheritance, false); - - auto link1 = new Link("Expression", {inheritance_handle, human_handle, mammal_handle}); - auto link2 = new Link("Expression", {inheritance_handle, monkey_handle, mammal_handle}); - string link1_handle = backend->add_link(link1, false); - string link2_handle = backend->add_link(link2, false); - backend->re_index_patterns(true); - return {link1_handle, link2_handle}; -} - TEST(RemoteAtomDBFederationTest, MetadataAggregationFromNestedPeer) { auto backend = make_shared("fed_nested_backend_"); - auto handles = populate_inheritance_links(backend); + auto handles = populate_inheritance_mammal_links(backend); map> peers; peers["nested"] = make_shared(backend, nullptr, "nested"); @@ -612,8 +761,8 @@ TEST(RemoteAtomDBFederationTest, MixedPeersDowngradeAndDeduplicate) { // Both contain the SAME two links so the facade must dedup by handle. auto nested_backend = make_shared("fed_mixed_nested_"); auto plain_backend = make_shared("fed_mixed_plain_"); - auto nested_handles = populate_inheritance_links(nested_backend); - auto plain_handles = populate_inheritance_links(plain_backend); + auto nested_handles = populate_inheritance_mammal_links(nested_backend); + auto plain_handles = populate_inheritance_mammal_links(plain_backend); ASSERT_EQ(nested_handles, plain_handles); // identical content -> identical handles map> peers; @@ -696,12 +845,12 @@ TEST(RemoteAtomDBFederationTest, CacheFirstProbingAcrossPeers) { // Before any read, nothing is cached. EXPECT_EQ(peer2->get_cached_atom(handle), nullptr); - // Phase 1 (cache) misses everywhere; Phase 2 escalation resolves from peer2's backend. + // Readonly peers: cache miss then escalate to peer2's backend. auto first = db->get_atom(handle); ASSERT_NE(first, nullptr); EXPECT_EQ(first->handle(), handle); - // peer2 must have warmed its cache, so a subsequent Phase 1 probe hits. + // peer2 must have warmed its cache, so a subsequent probe hits. EXPECT_NE(peer2->get_cached_atom(handle), nullptr); auto second = db->get_atom(handle); @@ -712,6 +861,107 @@ TEST(RemoteAtomDBFederationTest, CacheFirstProbingAcrossPeers) { EXPECT_EQ(db->get_atom("ffffffffffffffffffffffffffffffff"), nullptr); } +TEST(RemoteAtomDBFederationTest, PersistLinkWithoutCrossPeerTargetCopy) { + // peer3 writable backend + local_persistence start empty; targets live only on peer1. + // InMemoryDB (and RedisMongoDB with composite_type_enabled=false) can still persist the + // staged link itself. + auto peer1_remote = make_shared("persist_peer1_remote_"); + auto peer3_remote = make_shared("persist_peer3_remote_"); + auto peer3_local = make_shared("persist_peer3_local_"); + + auto a = new Node("Symbol", "\"persistA\""); + auto b = new Node("Symbol", "\"persistB\""); + auto implication = new Node("Symbol", "Implication"); + string a_h = peer1_remote->add_node(a, false); + string b_h = peer1_remote->add_node(b, false); + string impl_h = peer1_remote->add_node(implication, false); + + map> peers; + peers["peer1"] = make_shared(peer1_remote, nullptr, "peer1"); + peers["peer3"] = make_shared(peer3_remote, peer3_local, "peer3"); + auto db = make_shared(peers); + + auto link = make_shared( + "Expression", vector{impl_h, a_h, b_h}, true, Properties{{"strength", 0.833333}}); + string handle = db->add_link(link.get(), false); + ASSERT_FALSE(handle.empty()); + + EXPECT_FALSE(peer3_local->link_exists(handle)); + EXPECT_FALSE(peer3_local->atom_exists(a_h)); + EXPECT_FALSE(peer3_local->atom_exists(b_h)); + EXPECT_FALSE(peer3_local->atom_exists(impl_h)); + + db->get_peer("peer3")->release_cache(true, false); + + ASSERT_TRUE(peer3_local->link_exists(handle)); + EXPECT_FALSE(peer3_local->atom_exists(a_h)); + EXPECT_FALSE(peer3_local->atom_exists(b_h)); + EXPECT_FALSE(peer3_local->atom_exists(impl_h)); + + auto persisted = peer3_local->get_atom(handle); + ASSERT_NE(persisted, nullptr); + EXPECT_DOUBLE_EQ(persisted->custom_attributes.get_or("strength", -1.0), 0.833333); + + // Fresh facade sharing peer3 local persistence still sees the updated link. + map> reader_peers; + reader_peers["peer1"] = make_shared(peer1_remote, nullptr, "peer1"); + reader_peers["peer3"] = make_shared(peer3_remote, peer3_local, "peer3"); + auto reader = make_shared(reader_peers); + auto from_reader = reader->get_atom(handle); + ASSERT_NE(from_reader, nullptr); + EXPECT_DOUBLE_EQ(from_reader->custom_attributes.get_or("strength", -1.0), 0.833333); +} + +TEST(RemoteAtomDBFederationTest, StrengthUpdateVisibleAcrossPeers) { + // Writable peer3 local_persistence gets a strength update for a content-addressed handle. + // Readonly peers may still serve a stale cached/remote copy; facade must prefer peer3. + auto peer1_remote = make_shared("strength_peer1_remote_"); + auto peer2_remote = make_shared("strength_peer2_remote_"); + auto peer3_remote = make_shared("strength_peer3_remote_"); + auto peer3_local = make_shared("strength_peer3_local_"); + + auto a = new Node("Symbol", "\"A\""); + auto b = new Node("Symbol", "\"B\""); + auto implication = new Node("Symbol", "Implication"); + string a_h = peer1_remote->add_node(a, false); + string b_h = peer1_remote->add_node(b, false); + string impl_h = peer1_remote->add_node(implication, false); + peer2_remote->add_node(a, false); + peer2_remote->add_node(b, false); + peer2_remote->add_node(implication, false); + + auto weak = make_shared( + "Expression", vector{impl_h, a_h, b_h}, true, Properties{{"strength", 0.833333}}); + string handle = weak->handle(); + peer1_remote->add_link(weak.get(), false); + + map> peers; + peers["peer1"] = make_shared(peer1_remote, nullptr, "peer1"); + peers["peer2"] = make_shared(peer2_remote, nullptr, "peer2"); + peers["peer3"] = make_shared(peer3_remote, peer3_local, "peer3"); + auto db = make_shared(peers); + + // Warm readonly peer1 cache with the weak strength. + auto first = db->get_atom(handle); + ASSERT_NE(first, nullptr); + EXPECT_DOUBLE_EQ(first->custom_attributes.get_or("strength", -1.0), 0.833333); + EXPECT_NE(db->get_peer("peer1")->get_cached_atom(handle), nullptr); + + // Another process persists a stronger copy into shared peer3 local_persistence. + auto strong = make_shared( + "Expression", vector{impl_h, a_h, b_h}, true, Properties{{"strength", 0.942654}}); + ASSERT_EQ(strong->handle(), handle); + peer3_local->add_node(a, false); + peer3_local->add_node(b, false); + peer3_local->add_node(implication, false); + peer3_local->add_link(strong.get(), false); + + // Same process still has stale peer1 cache; must still read the updated strength. + auto updated = db->get_atom(handle); + ASSERT_NE(updated, nullptr); + EXPECT_DOUBLE_EQ(updated->custom_attributes.get_or("strength", -1.0), 0.942654); +} + int main(int argc, char** argv) { ::testing::InitGoogleTest(&argc, argv); return RUN_ALL_TESTS(); diff --git a/src/tests/main/evaluation_evolution.cc b/src/tests/main/evaluation_evolution.cc index 2cea63d7..4597aec5 100644 --- a/src/tests/main/evaluation_evolution.cc +++ b/src/tests/main/evaluation_evolution.cc @@ -19,6 +19,7 @@ #include "PatternMatchingQueryProxy.h" #include "QueryAnswer.h" #include "QueryEvolutionProxy.h" +#include "RemoteAtomDB.h" #include "ServiceBusSingleton.h" #include "SystemParametersSingleton.h" #include "TestAtomDBJsonConfig.h" @@ -292,6 +293,14 @@ static shared_ptr issue_context_creation_query( return proxy; } +static void flush_remote_link_template_cache(bool force = false) { + if (auto remote_db = dynamic_pointer_cast(db); remote_db != nullptr) { + auto link_schema = + LinkSchema({LINK_TEMPLATE, EXPRESSION, "3", VARIABLE, "V1", VARIABLE, "V2", VARIABLE, "V3"}); + remote_db->release_caches(link_schema, true, force); + } +} + static void insert_or_update(map& count_map, const string& key, double value) { STACK_TRACE(); auto iterator = count_map.find(key); @@ -813,6 +822,8 @@ static void build_links(const vector& query, if (!proxy->finished()) { proxy->abort(); } + // Persist staged writes once per build cycle instead of per created link. + flush_remote_link_template_cache(); LOG_DEBUG("Built " + to_string(count_created) + " links"); } @@ -992,7 +1003,12 @@ static void add_preset_links(const vector& implication_to_target_predica auto link = std::dynamic_pointer_cast(parser_handler->element_stack.top()); link->custom_attributes["strength"] = (double) Utils::string_to_float(line[0]); LOG_DEBUG("Adding Link: [" + line[0] + "] " + line[1]); - db->add_link(link.get(), false); + vector atoms_to_add; + atoms_to_add.reserve(parser_handler->handle_to_atom.size()); + for (const auto& [_, atom] : parser_handler->handle_to_atom) { + atoms_to_add.push_back(atom.get()); + } + db->add_atoms(atoms_to_add, false, true); count++; line.clear(); buffer_determiners.push_back({link->handle(), link->targets[1], link->targets[2]}); @@ -1018,6 +1034,7 @@ static void add_preset_links(const vector& implication_to_target_predica LOG_INFO("Updating determiners in AttentionBroker"); AttentionBrokerClient::set_determiners(buffer_determiners, context); buffer_determiners.clear(); + flush_remote_link_template_cache(); } static void run(const string& context_tag) { @@ -1393,6 +1410,7 @@ static void insert_type_symbols() { db->add_node(node, false); delete (node); } + flush_remote_link_template_cache(); } int main(int argc, char* argv[]) { @@ -1500,5 +1518,7 @@ int main(int argc, char* argv[]) { LOG_INFO("FINAL_RESULT " + to_string(best_strength) + " " + to_string(iteration) + " " + Utils::join(test_label, '_') + " " + answer_to_string(recorded_answers.back().first)); + flush_remote_link_template_cache(true); + return 0; }