feat(clp-s)!: Deduplicate cached search results using compound MongoDB IDs.

components/clp-mcp-server/clp_mcp_server/clp_connector.py

@@ -161,11 +161,14 @@ async def read_results(self, query_id: str) -> list[dict]:
         results = []
 
         async for doc in collection.find({}, limit=SEARCH_MAX_NUM_RESULTS):
+            result_id = doc["_id"]
+            doc["archive_id"] = result_id["archive_id"]
+            doc["log_event_idx"] = result_id["log_event_idx"]
             doc["link"] = (
                 f"{self._webui_addr}/streamFile?type=json"
                 f"&streamId={doc['archive_id']}"
                 f"&dataset={CLP_DEFAULT_DATASET_NAME}"
-                f"&logEventIdx={doc['log_event_ix']}"
+                f"&logEventIdx={doc['log_event_idx']}"
             )
             doc["_id"] = None
             results.append(doc)

components/clp-mcp-server/tests/test_clp_connector.py

@@ -111,9 +111,9 @@ async def test_read_results_returns_docs(mock_clp_config: Any) -> None:
     """Tests reading results returns expected documents."""
     connector = ClpConnector(mock_clp_config)
     mock_docs = [
-        {"_id": "1", "archive_id": "archA", "log_event_ix": 1},
-        {"_id": "2", "archive_id": "archB", "log_event_ix": 2},
-        {"_id": "3", "archive_id": "archC", "log_event_ix": 3},
+        {"_id": {"archive_id": "archA", "log_event_idx": 1}},
+        {"_id": {"archive_id": "archB", "log_event_idx": 2}},
+        {"_id": {"archive_id": "archC", "log_event_idx": 3}},
     ]
     mock_collection = AsyncMock()
     mock_collection.find = MagicMock(return_value=_aiter(mock_docs))
@@ -129,8 +129,8 @@ async def test_read_results_adds_link_field(mock_clp_config: Any) -> None:
     """Ensures read_results adds a 'link' field."""
     connector = ClpConnector(mock_clp_config)
     mock_docs = [
-        {"_id": "1", "archive_id": "archA", "log_event_ix": 10},
-        {"_id": "2", "archive_id": "archB", "log_event_ix": 20},
+        {"_id": {"archive_id": "archA", "log_event_idx": 10}},
+        {"_id": {"archive_id": "archB", "log_event_idx": 20}},
     ]
     mock_collection = AsyncMock()
     mock_collection.find = MagicMock(return_value=_aiter(mock_docs))
@@ -143,7 +143,7 @@ async def test_read_results_adds_link_field(mock_clp_config: Any) -> None:
         expected_link = (
             f"http://{mock_clp_config.webui.host}:{mock_clp_config.webui.port}"
             f"/streamFile?type=json&streamId={original['archive_id']}"
-            f"&dataset=default&logEventIdx={original['log_event_ix']}"
+            f"&dataset=default&logEventIdx={original['log_event_idx']}"
         )
         assert result["link"] == expected_link
 

components/core/src/clp/clo/CMakeLists.txt

@@ -166,8 +166,14 @@ set(
         ../../reducer/types.hpp
 )
 
+set(
+        CLP_S_MONGODB_UTILS_SOURCES
+        ../../clp_s/MongoDBUtils.cpp
+        ../../clp_s/MongoDBUtils.hpp
+)
+
 if(CLP_BUILD_EXECUTABLES)
-        add_executable(clo ${CLO_SOURCES} ${REDUCER_SOURCES})
+        add_executable(clo ${CLO_SOURCES} ${CLP_S_MONGODB_UTILS_SOURCES} ${REDUCER_SOURCES})
         target_compile_features(clo PRIVATE cxx_std_20)
         target_include_directories(clo
                 PRIVATE

components/core/src/clp/clo/OutputHandler.cpp

@@ -4,9 +4,12 @@
 #include <string>
 #include <string_view>
 
+#include <mongocxx/exception/bulk_write_exception.hpp>
 #include <msgpack.hpp>
 #include <spdlog/spdlog.h>
 
+#include <clp_s/MongoDBUtils.hpp>
+
 #include "../../reducer/CountOperator.hpp"
 #include "../../reducer/network_utils.hpp"
 #include "../networking/socket_utils.hpp"
@@ -51,6 +54,7 @@ ResultsCacheOutputHandler::ResultsCacheOutputHandler(
 )
         : m_batch_size(batch_size),
           m_max_num_results(max_num_results) {
+    m_insert_options.ordered(false);
     try {
         auto mongo_uri = mongocxx::uri(uri);
         m_client = mongocxx::client(mongo_uri);
@@ -106,17 +110,24 @@ ErrorCode ResultsCacheOutputHandler::flush() {
                     std::move(
                             bsoncxx::builder::basic::make_document(
                                     bsoncxx::builder::basic::kvp(
-                                            cResultsCacheKeys::SearchOutput::OrigFileId,
-                                            std::move(result.orig_file_id)
+                                            cResultsCacheKeys::SearchOutput::Id,
+                                            bsoncxx::builder::basic::make_document(
+                                                    bsoncxx::builder::basic::kvp(
+                                                            cResultsCacheKeys::SearchOutput::
+                                                                    OrigFileId,
+                                                            result.orig_file_id
+                                                    ),
+                                                    bsoncxx::builder::basic::kvp(
+                                                            cResultsCacheKeys::SearchOutput::
+                                                                    LogEventIdx,
+                                                            result.log_event_ix
+                                                    )
+                                            )
                                     ),
                                     bsoncxx::builder::basic::kvp(
                                             cResultsCacheKeys::SearchOutput::OrigFilePath,
                                             std::move(result.orig_file_path)
                                     ),
-                                    bsoncxx::builder::basic::kvp(
-                                            cResultsCacheKeys::SearchOutput::LogEventIx,
-                                            result.log_event_ix
-                                    ),
                                     bsoncxx::builder::basic::kvp(
                                             cResultsCacheKeys::SearchOutput::Timestamp,
                                             result.timestamp
@@ -128,29 +139,42 @@ ErrorCode ResultsCacheOutputHandler::flush() {
                             )
                     )
             );
-            count++;
-
-            if (count == m_batch_size) {
-                m_collection.insert_many(m_results);
-                m_results.clear();
-                count = 0;
-            }
         } catch (mongocxx::exception const& e) {
+            SPDLOG_ERROR("Failed to build search result - {}", e.what());
             return ErrorCode::ErrorCode_Failure_DB_Bulk_Write;
         }
-    }
 
-    try {
-        if (false == m_results.empty()) {
-            m_collection.insert_many(m_results);
-            m_results.clear();
+        count++;
+        if (count == m_batch_size) {
+            if (false == insert_results()) {
+                return ErrorCode::ErrorCode_Failure_DB_Bulk_Write;
+            }
+            count = 0;
         }
-    } catch (mongocxx::exception const& e) {
+    }
+
+    if (false == m_results.empty() && false == insert_results()) {
         return ErrorCode::ErrorCode_Failure_DB_Bulk_Write;
     }
     return ErrorCode::ErrorCode_Success;
 }
 
+auto ResultsCacheOutputHandler::insert_results() -> bool {
+    try {
+        m_collection.insert_many(m_results, m_insert_options);
+    } catch (mongocxx::bulk_write_exception const& exception) {
+        if (false == clp_s::contains_only_duplicate_key_write_errors(exception, m_results.size())) {
+            SPDLOG_ERROR("Failed to insert search results - {}", exception.what());
+            return false;
+        }
+    } catch (mongocxx::exception const& exception) {
+        SPDLOG_ERROR("Failed to insert search results - {}", exception.what());
+        return false;
+    }
+    m_results.clear();
+    return true;
+}
+
 CountOutputHandler::CountOutputHandler(int reducer_socket_fd)
         : m_reducer_socket_fd{reducer_socket_fd},
           m_pipeline{reducer::PipelineInputMode::InterStage} {

components/core/src/clp/clo/OutputHandler.hpp

@@ -10,6 +10,7 @@
 #include <mongocxx/client.hpp>
 #include <mongocxx/collection.hpp>
 #include <mongocxx/exception/exception.hpp>
+#include <mongocxx/options/insert.hpp>
 #include <mongocxx/uri.hpp>
 
 #include "../../reducer/Pipeline.hpp"
@@ -208,8 +209,16 @@ class ResultsCacheOutputHandler : public OutputHandler {
         return m_latest_results.size() >= m_max_num_results;
     }
 
+    /**
+     * Inserts the pending results as an unordered batch. Duplicate-key errors are treated as
+     * success so that retries converge on the complete result set.
+     * @return Whether insertion succeeded or produced only duplicate-key errors.
+     */
+    [[nodiscard]] auto insert_results() -> bool;
+
     mongocxx::client m_client;
     mongocxx::collection m_collection;
+    mongocxx::options::insert m_insert_options;
     std::vector<bsoncxx::document::value> m_results;
     uint64_t m_batch_size;
     uint64_t m_max_num_results;

components/core/src/clp/clo/constants.hpp

@@ -12,9 +12,10 @@ constexpr char IsLastChunk[]{"is_last_chunk"};
 }  // namespace IrOutput
 
 namespace SearchOutput {
+constexpr char Id[]{"_id"};
 constexpr char OrigFileId[]{"orig_file_id"};
 constexpr char OrigFilePath[]{"orig_file_path"};
-constexpr char LogEventIx[]{"log_event_ix"};
+constexpr char LogEventIdx[]{"log_event_idx"};
 constexpr char Timestamp[]{"timestamp"};
 constexpr char Message[]{"message"};
 }  // namespace SearchOutput

components/core/src/clp_s/CMakeLists.txt

@@ -492,6 +492,8 @@ set(
         ErrorCode.hpp
         kv_ir_search.cpp
         kv_ir_search.hpp
+        MongoDBUtils.cpp
+        MongoDBUtils.hpp
         OutputHandlerImpl.cpp
         OutputHandlerImpl.hpp
         ResultsCacheUtils.cpp

components/core/src/clp_s/MongoDBUtils.cpp

@@ -0,0 +1,128 @@
+#include "MongoDBUtils.hpp"
+
+#include <cstdint>
+
+#include <bsoncxx/types.hpp>
+#include <mongocxx/exception/bulk_write_exception.hpp>
+
+namespace clp_s {
+namespace {
+constexpr int32_t cDuplicateKeyErrorCode{11'000};
+
+/**
+ * Checks whether an aggregated bulk-write reply contains any command errors.
+ * @param reply The raw MongoDB bulk-write reply.
+ * @return Whether the reply contains a command error.
+ */
+[[nodiscard]] auto has_command_errors(bsoncxx::document::view const& reply) -> bool;
+
+/**
+ * Checks whether a bulk-write reply contains any write-concern errors.
+ * @param reply The raw MongoDB bulk-write reply.
+ * @return Whether the reply contains a write-concern error.
+ */
+[[nodiscard]] auto has_write_concern_errors(bsoncxx::document::view const& reply) -> bool;
+
+/**
+ * Checks whether an entry from a bulk-write reply's `writeErrors` array is a duplicate-key error.
+ * @param write_error The write-error entry to inspect.
+ * @return Whether the entry has MongoDB's duplicate-key error code.
+ */
+[[nodiscard]] auto is_duplicate_key_write_error(bsoncxx::array::element const& write_error) -> bool;
+
+[[nodiscard]] auto has_command_errors(bsoncxx::document::view const& reply) -> bool {
+    auto const errors_element = reply["errorReplies"];
+    if (false == static_cast<bool>(errors_element)) {
+        return false;
+    }
+    if (bsoncxx::type::k_array != errors_element.type()) {
+        return true;
+    }
+    auto const errors = errors_element.get_array().value;
+    return false == errors.empty();
+}
+
+[[nodiscard]] auto has_write_concern_errors(bsoncxx::document::view const& reply) -> bool {
+    if (static_cast<bool>(reply["writeConcernError"])) {
+        return true;
+    }
+
+    auto const errors_element = reply["writeConcernErrors"];
+    if (false == static_cast<bool>(errors_element)) {
+        return false;
+    }
+    if (bsoncxx::type::k_array != errors_element.type()) {
+        return true;
+    }
+    auto const errors = errors_element.get_array().value;
+    return false == errors.empty();
+}
+
+[[nodiscard]] auto is_duplicate_key_write_error(bsoncxx::array::element const& write_error)
+        -> bool {
+    if (bsoncxx::type::k_document != write_error.type()) {
+        return false;
+    }
+
+    auto const code = write_error.get_document().value["code"];
+    if (false == static_cast<bool>(code)) {
+        return false;
+    }
+    if (bsoncxx::type::k_int32 == code.type()) {
+        return cDuplicateKeyErrorCode == code.get_int32().value;
+    }
+    if (bsoncxx::type::k_int64 == code.type()) {
+        return cDuplicateKeyErrorCode == code.get_int64().value;
+    }
+    return false;
+}
+}  // namespace
+
+auto contains_only_duplicate_key_write_errors(
+        mongocxx::bulk_write_exception const& exception,
+        size_t num_documents
+) -> bool {
+    auto const& raw_server_error = exception.raw_server_error();
+    if (false == raw_server_error.has_value()) {
+        return false;
+    }
+
+    auto const reply = raw_server_error->view();
+    if (has_command_errors(reply) || has_write_concern_errors(reply)) {
+        return false;
+    }
+
+    auto const write_errors_element = reply["writeErrors"];
+    if (false == static_cast<bool>(write_errors_element)
+        || bsoncxx::type::k_array != write_errors_element.type())
+    {
+        return false;
+    }
+
+    auto const write_errors = write_errors_element.get_array().value;
+    if (write_errors.empty()) {
+        return false;
+    }
+
+    size_t num_write_errors{0};
+    for (auto const& write_error : write_errors) {
+        if (false == is_duplicate_key_write_error(write_error)) {
+            return false;
+        }
+        ++num_write_errors;
+    }
+
+    auto const num_inserted_element = reply["nInserted"];
+    if (false == static_cast<bool>(num_inserted_element)
+        || bsoncxx::type::k_int32 != num_inserted_element.type())
+    {
+        return false;
+    }
+    auto const num_inserted = num_inserted_element.get_int32().value;
+    if (num_inserted < 0) {
+        return false;
+    }
+
+    return static_cast<size_t>(num_inserted) + num_write_errors == num_documents;
+}
+}  // namespace clp_s

components/core/src/clp_s/MongoDBUtils.hpp

@@ -0,0 +1,26 @@
+#ifndef CLP_S_MONGODBUTILS_HPP
+#define CLP_S_MONGODBUTILS_HPP
+
+#include <cstddef>
+
+#include <mongocxx/exception/bulk_write_exception.hpp>
+
+namespace clp_s {
+/**
+ * Returns whether the bulk write failed only because some documents already exist.
+ *
+ * Command and write-concern errors are rejected since they mean MongoDB did not confirm the outcome
+ * of the entire batch. The number of inserted documents and write errors must account for every
+ * submitted document, and every write error must be a duplicate-key error.
+ * @param exception The exception containing the raw MongoDB bulk-write reply.
+ * @param num_documents The number of documents submitted in the bulk write.
+ * @return Whether the reply accounts for every document using successful inserts and duplicate-key
+ * errors only.
+ */
+[[nodiscard]] auto contains_only_duplicate_key_write_errors(
+        mongocxx::bulk_write_exception const& exception,
+        size_t num_documents
+) -> bool;
+}  // namespace clp_s
+
+#endif  // CLP_S_MONGODBUTILS_HPP