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

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

@@ -160,12 +160,19 @@ 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"]
+            if isinstance(result_id, dict):
+                doc["archive_id"] = result_id["archive_id"]
+                doc["log_event_idx"] = result_id["log_event_idx"]
+            else:
+                doc["log_event_idx"] = doc.pop("log_event_ix")
             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

@@ -110,27 +110,45 @@ async def test_wait_query_completion_failure_cases(
 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": {"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))
+
+    with patch.object(connector, "_results_cache", {"12": mock_collection}):
+        results = await connector.read_results("12")
+
+    assert len(results) == len(mock_docs)
+
+
+@pytest.mark.asyncio
+async def test_read_results_supports_legacy_docs(mock_clp_config: Any) -> None:
+    """Tests reading legacy results with top-level archive and log-event fields."""
+    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},
     ]
     mock_collection = AsyncMock()
     mock_collection.find = MagicMock(return_value=_aiter(mock_docs))
 
     with patch.object(connector, "_results_cache", {"12": mock_collection}):
         results = await connector.read_results("12")
 
-    assert len(results) == len(mock_docs)
+    assert [result["archive_id"] for result in results] == ["archA", "archB"]
+    assert [result["log_event_idx"] for result in results] == [1, 2]
 
 
 @pytest.mark.asyncio
 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 +160,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_s/OutputHandlerImpl.cpp

@@ -1,16 +1,20 @@
 #include "OutputHandlerImpl.hpp"
 
+#include <algorithm>
 #include <sstream>
 #include <string>
 #include <string_view>
 #include <vector>
 
 #include <bsoncxx/builder/basic/document.hpp>
 #include <bsoncxx/builder/basic/kvp.hpp>
+#include <bsoncxx/types.hpp>
 #include <mongocxx/client.hpp>
 #include <mongocxx/collection.hpp>
+#include <mongocxx/exception/bulk_write_exception.hpp>
 #include <mongocxx/exception/exception.hpp>
 #include <mongocxx/instance.hpp>
+#include <mongocxx/options/insert.hpp>
 #include <msgpack.hpp>
 #include <spdlog/spdlog.h>
 
@@ -28,6 +32,117 @@ using std::string;
 using std::string_view;
 
 namespace clp_s {
+namespace {
+constexpr int32_t cDuplicateKeyErrorCode{11'000};
+
+/**
+ * Checks whether a bulk-write reply reports a successful command.
+ * @param reply The raw MongoDB bulk-write reply.
+ * @return true if the reply contains no command error, false otherwise.
+ */
+[[nodiscard]] auto is_successful_command_reply(bsoncxx::document::view const& reply) -> bool {
+    if (static_cast<bool>(reply["code"]) || static_cast<bool>(reply["errmsg"])) {
+        return false;
+    }
+
+    auto const command_status = reply["ok"];
+    if (false == static_cast<bool>(command_status)) {
+        return true;
+    }
+    if (bsoncxx::type::k_double == command_status.type()) {
+        return 1.0 == command_status.get_double().value;
+    }
+    if (bsoncxx::type::k_int32 == command_status.type()) {
+        return 1 == command_status.get_int32().value;
+    }
+    if (bsoncxx::type::k_int64 == command_status.type()) {
+        return 1 == command_status.get_int64().value;
+    }
+    return false;
+}
+
+/**
+ * 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 {
+    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();
+}
+
+/**
+ * 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 {
+    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;
+}
+
+/**
+ * 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 {
+    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 (false == is_successful_command_reply(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
+
 void FileOutputHandler::write(
         string_view message,
         epochtime_t timestamp,
@@ -96,6 +211,21 @@ ErrorCode ResultsCacheOutputHandler::finish() {
             m_results.emplace_back(
                     std::move(
                             bsoncxx::builder::basic::make_document(
+                                    bsoncxx::builder::basic::kvp(
+                                            constants::results_cache::search::cId,
+                                            bsoncxx::builder::basic::make_document(
+                                                    bsoncxx::builder::basic::kvp(
+                                                            constants::results_cache::search::
+                                                                    cArchiveId,
+                                                            result.archive_id
+                                                    ),
+                                                    bsoncxx::builder::basic::kvp(
+                                                            constants::results_cache::search::
+                                                                    cLogEventIdx,
+                                                            result.log_event_idx
+                                                    )
+                                            )
+                                    ),
                                     bsoncxx::builder::basic::kvp(
                                             constants::results_cache::search::cOrigFilePath,
                                             std::move(result.original_path)
@@ -108,14 +238,6 @@ ErrorCode ResultsCacheOutputHandler::finish() {
                                             constants::results_cache::search::cTimestamp,
                                             result.timestamp
                                     ),
-                                    bsoncxx::builder::basic::kvp(
-                                            constants::results_cache::search::cArchiveId,
-                                            std::move(result.archive_id)
-                                    ),
-                                    bsoncxx::builder::basic::kvp(
-                                            constants::results_cache::search::cLogEventIx,
-                                            result.log_event_idx
-                                    ),
                                     bsoncxx::builder::basic::kvp(
                                             std::string{constants::results_cache::search::cDataset},
                                             std::move(result.dataset)
@@ -126,21 +248,25 @@ ErrorCode ResultsCacheOutputHandler::finish() {
             count++;
 
             if (count == m_batch_size) {
-                m_collection.insert_many(m_results);
-                m_results.clear();
+                if (false == insert_results()) {
+                    return ErrorCode::ErrorCodeFailureDbBulkWrite;
+                }
                 count = 0;
             }
         } catch (mongocxx::exception const& e) {
+            SPDLOG_ERROR("Failed to build or insert search results - {}", e.what());
             return ErrorCode::ErrorCodeFailureDbBulkWrite;
         }
     }
 
     try {
         if (false == m_results.empty()) {
-            m_collection.insert_many(m_results);
-            m_results.clear();
+            if (false == insert_results()) {
+                return ErrorCode::ErrorCodeFailureDbBulkWrite;
+            }
         }
     } catch (mongocxx::exception const& e) {
+        SPDLOG_ERROR("Failed to insert final search-results batch - {}", e.what());
         return ErrorCode::ErrorCodeFailureDbBulkWrite;
     }
     return ErrorCode::ErrorCodeSuccess;
@@ -178,6 +304,24 @@ void ResultsCacheOutputHandler::write(
     }
 }
 
+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 == 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;
+}
+
 CountReducerOutputHandler::CountReducerOutputHandler(int reducer_socket_fd)
         : search::OutputHandler(false, false),
           m_reducer_socket_fd(reducer_socket_fd),

components/core/src/clp_s/OutputHandlerImpl.hpp

@@ -200,6 +200,13 @@ class ResultsCacheOutputHandler : public ::clp_s::search::OutputHandler {
     void write(std::string_view message) override { write(message, 0, {}, 0); }
 
 private:
+    /**
+     * 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_s/archive_constants.hpp

@@ -53,8 +53,9 @@ constexpr char cIsLastChunk[]{"is_last_chunk"};
 }  // namespace results_cache::decompression
 
 namespace results_cache::search {
+constexpr char cId[]{"_id"};
 constexpr char cOrigFilePath[]{"orig_file_path"};
-constexpr char cLogEventIx[]{"log_event_ix"};
+constexpr char cLogEventIdx[]{"log_event_idx"};
 constexpr char cTimestamp[]{"timestamp"};
 constexpr char cMessage[]{"message"};
 constexpr char cArchiveId[]{"archive_id"};