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

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

@@ -160,12 +160,16 @@ async def read_results(self, query_id: str) -> list[dict]:
         collection = self._results_cache[str(query_id)]
         results = []
 
-        async for doc in collection.find({}, limit=SEARCH_MAX_NUM_RESULTS):
+        async for raw_doc in collection.find({}, limit=SEARCH_MAX_NUM_RESULTS):
+            doc = dict(raw_doc)
+            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))
@@ -142,8 +142,8 @@ async def test_read_results_adds_link_field(mock_clp_config: Any) -> None:
     for original, result in zip(mock_docs, results, strict=True):
         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"/streamFile?type=json&streamId={original['_id']['archive_id']}"
+            f"&dataset=default&logEventIdx={original['_id']['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,13 @@
 #include <string>
 #include <string_view>
 
+#include <mongocxx/exception/bulk_write_exception.hpp>
+#include <mongocxx/options/insert.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"
@@ -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,44 @@ 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 {
+        mongocxx::options::insert options;
+        options.ordered(false);
+        m_collection.insert_many(m_results, options);
+    } catch (mongocxx::bulk_write_exception const& exception) {
+        if (false == clp_s::contains_only_duplicate_key_write_errors(exception)) {
+            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

@@ -208,6 +208,13 @@ 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 true on success or duplicate-key-only errors, false otherwise.
+     */
+    [[nodiscard]] auto insert_results() -> bool;
+
     mongocxx::client m_client;
     mongocxx::collection m_collection;
     std::vector<bsoncxx::document::value> m_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,106 @@
+#include "MongoDBUtils.hpp"
+
+#include <algorithm>
+
+#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 true if the reply contains a command error, false otherwise.
+ */
+[[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 true if the reply contains a write-concern error, false otherwise.
+ */
+[[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 true if the entry has MongoDB's duplicate-key error code, false otherwise.
+ */
+[[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 errors.begin() != errors.end();
+}
+
+[[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 errors.begin() != errors.end();
+}
+
+[[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)
+        -> 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.begin() == write_errors.end()) {
+        return false;
+    }
+    return std::all_of(write_errors.begin(), write_errors.end(), is_duplicate_key_write_error);
+}
+}  // namespace clp_s

components/core/src/clp_s/MongoDBUtils.hpp

@@ -0,0 +1,21 @@
+#ifndef CLP_S_MONGODBUTILS_HPP
+#define CLP_S_MONGODBUTILS_HPP
+
+#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. At least one write error must be present, and every write error
+ * must be a duplicate-key error.
+ * @param exception The exception containing the raw MongoDB bulk-write reply.
+ * @return true if the reply contains only duplicate-key write errors, false otherwise.
+ */
+[[nodiscard]] auto contains_only_duplicate_key_write_errors(
+        mongocxx::bulk_write_exception const& exception
+) -> bool;
+}  // namespace clp_s
+
+#endif  // CLP_S_MONGODBUTILS_HPP