From 767feb1f04b19e1216e7997a0beea4913f7e50f2 Mon Sep 17 00:00:00 2001 From: Sitao Wang Date: Mon, 31 Aug 2026 12:44:51 -0400 Subject: [PATCH 01/27] Insert compound id in mongodb --- .../core/src/clp_s/OutputHandlerImpl.cpp | 92 ++++++++++++++++++- .../core/src/clp_s/OutputHandlerImpl.hpp | 7 ++ .../core/src/clp_s/archive_constants.hpp | 1 + 3 files changed, 96 insertions(+), 4 deletions(-) diff --git a/components/core/src/clp_s/OutputHandlerImpl.cpp b/components/core/src/clp_s/OutputHandlerImpl.cpp index e2ebbe48f2..673d7022e2 100644 --- a/components/core/src/clp_s/OutputHandlerImpl.cpp +++ b/components/core/src/clp_s/OutputHandlerImpl.cpp @@ -7,10 +7,13 @@ #include #include +#include #include #include +#include #include #include +#include #include #include @@ -28,6 +31,50 @@ using std::string; using std::string_view; namespace clp_s { +namespace { +constexpr int32_t cDuplicateKeyErrorCode{11'000}; + +[[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 write_errors_element = raw_server_error->view()["writeErrors"]; + if (false == static_cast(write_errors_element) + || bsoncxx::type::k_array != write_errors_element.type()) + { + return false; + } + + bool found_write_error{false}; + for (auto const& write_error_element : write_errors_element.get_array().value) { + if (bsoncxx::type::k_document != write_error_element.type()) { + return false; + } + auto const code_element = write_error_element.get_document().value["code"]; + if (false == static_cast(code_element)) { + return false; + } + if (bsoncxx::type::k_int32 == code_element.type()) { + if (cDuplicateKeyErrorCode != code_element.get_int32().value) { + return false; + } + } else if (bsoncxx::type::k_int64 == code_element.type()) { + if (cDuplicateKeyErrorCode != code_element.get_int64().value) { + return false; + } + } else { + return false; + } + found_write_error = true; + } + return found_write_error; +} +} // namespace + void FileOutputHandler::write( string_view message, epochtime_t timestamp, @@ -96,6 +143,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:: + cLogEventIx, + result.log_event_idx + ) + ) + ), bsoncxx::builder::basic::kvp( constants::results_cache::search::cOrigFilePath, std::move(result.original_path) @@ -126,26 +188,48 @@ 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; } +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; +} + void ResultsCacheOutputHandler::write( string_view message, epochtime_t timestamp, diff --git a/components/core/src/clp_s/OutputHandlerImpl.hpp b/components/core/src/clp_s/OutputHandlerImpl.hpp index 5cacff8966..cdaa24eee3 100644 --- a/components/core/src/clp_s/OutputHandlerImpl.hpp +++ b/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 m_results; diff --git a/components/core/src/clp_s/archive_constants.hpp b/components/core/src/clp_s/archive_constants.hpp index 0dbf3d4c68..6dfae883e7 100644 --- a/components/core/src/clp_s/archive_constants.hpp +++ b/components/core/src/clp_s/archive_constants.hpp @@ -53,6 +53,7 @@ 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 cTimestamp[]{"timestamp"}; From 3b4ad6566575b41c4141b9cc80904a9c4f6f224d Mon Sep 17 00:00:00 2001 From: Sitao Wang Date: Mon, 31 Aug 2026 14:15:49 -0400 Subject: [PATCH 02/27] Use query complete time for gargabe collection --- .../search_result_garbage_collector.py | 126 +++++++++++++----- 1 file changed, 91 insertions(+), 35 deletions(-) diff --git a/components/job-orchestration/job_orchestration/garbage_collector/search_result_garbage_collector.py b/components/job-orchestration/job_orchestration/garbage_collector/search_result_garbage_collector.py index 72c24e76f2..99e574485b 100644 --- a/components/job-orchestration/job_orchestration/garbage_collector/search_result_garbage_collector.py +++ b/components/job-orchestration/job_orchestration/garbage_collector/search_result_garbage_collector.py @@ -1,33 +1,76 @@ import asyncio -from typing import Final +from contextlib import closing +from typing import Any, cast, Final import pymongo import pymongo.database -from bson import ObjectId -from clp_py_utils.clp_config import ClpConfig, ResultsCache +from clp_py_utils.clp_config import ( + ClpConfig, + Database, + QUERY_JOBS_TABLE_NAME, + ResultsCache, +) from clp_py_utils.clp_logging import configure_logging, get_logger +from clp_py_utils.sql_adapter import SqlAdapter from job_orchestration.garbage_collector.constants import ( MIN_TO_SECONDS, SEARCH_RESULT_GARBAGE_COLLECTOR_NAME, ) -from job_orchestration.garbage_collector.utils import get_expiry_epoch_secs # Constants MONGODB_ID_KEY: Final[str] = "_id" +MAX_NUM_JOB_IDS_PER_QUERY: Final[int] = 1000 logger = get_logger(SEARCH_RESULT_GARBAGE_COLLECTOR_NAME) -def _get_latest_doc_timestamp(collection: pymongo.collection.Collection) -> int: - latest_doc = collection.find_one(sort=[(MONGODB_ID_KEY, pymongo.DESCENDING)]) - if latest_doc is None: - return 0 - - object_id = latest_doc[MONGODB_ID_KEY] - if isinstance(object_id, ObjectId): - return int(object_id.generation_time.timestamp()) - raise ValueError(f"{object_id} is not an ObjectID") +def _get_expired_job_ids( + database_config: Database, job_ids: list[int], retention_period_minutes: int +) -> list[int]: + """ + Filter query-job IDs by whether their retention periods have ended. + + MariaDB computes each query's completion time as `creation_time + duration`. A query-job ID is + included when the time since completion is greater than `retention_period_minutes`. + + :param database_config: Configuration for the orchestration database. + :param job_ids: Query-job IDs to filter. + :param retention_period_minutes: Length of the retention period following query completion, in + minutes. + :return: Query-job IDs completed more than `retention_period_minutes` ago. + """ + if len(job_ids) == 0: + return [] + + expired_job_ids: list[int] = [] + sql_adapter = SqlAdapter(database_config) + with ( + closing(sql_adapter.create_connection(True)) as db_conn, + closing(db_conn.cursor()) as db_cursor, + ): + for begin_ix in range(0, len(job_ids), MAX_NUM_JOB_IDS_PER_QUERY): + job_ids_batch = job_ids[begin_ix : begin_ix + MAX_NUM_JOB_IDS_PER_QUERY] + job_id_placeholders = ",".join(["%s"] * len(job_ids_batch)) + query = ( + f""" + SELECT id + FROM `{QUERY_JOBS_TABLE_NAME}` + WHERE id IN ({job_id_placeholders}) + AND TIMESTAMPADD( + MICROSECOND, + CAST(duration * 1000000 AS SIGNED), + creation_time + ) < TIMESTAMPADD(MINUTE, %s, CURRENT_TIMESTAMP(3)) + """ # noqa: S608 + ) + db_cursor.execute( + query, + [*job_ids_batch, -retention_period_minutes], + ) + rows = cast("list[tuple[int]]", db_cursor.fetchall()) + expired_job_ids.extend(row[0] for row in rows) + return expired_job_ids def _delete_result_metadata( @@ -38,46 +81,59 @@ def _delete_result_metadata( def _collect_and_sweep_expired_search_results( - result_cache_config: ResultsCache, results_metadata_collection_name: str -): - expiry_epoch = get_expiry_epoch_secs(result_cache_config.retention_period) + result_cache_config: ResultsCache, + database_config: Database, + results_metadata_collection_name: str, +) -> None: + """ + Remove search results whose query completion time is older than the retention cutoff. - logger.debug(f"Searching for search jobs finished before {expiry_epoch}.") - deleted_job_ids: list[int] = [] - with pymongo.MongoClient(result_cache_config.get_uri()) as results_cache_client: - results_cache_db = results_cache_client.get_default_database() - collection_names = results_cache_db.list_collection_names() - for job_id in collection_names: - if not job_id.isdigit(): - continue + Numeric MongoDB collection names are interpreted as query-job IDs. Collections selected by + `_get_expired_job_ids` are dropped along with their result metadata documents. - job_results_collection = results_cache_db.get_collection(job_id) - collection_timestamp = _get_latest_doc_timestamp(job_results_collection) - if collection_timestamp >= expiry_epoch: - continue + :param result_cache_config: MongoDB result-cache and retention configuration. + :param database_config: Configuration for the orchestration database. + :param results_metadata_collection_name: Name of the result metadata collection. + """ + retention_period = result_cache_config.retention_period + if retention_period is None: + return - _delete_result_metadata(results_cache_db, results_metadata_collection_name, job_id) - job_results_collection.drop() - deleted_job_ids.append(int(job_id)) + deleted_job_ids: list[int] = [] + results_cache_client: pymongo.MongoClient[dict[str, Any]] = pymongo.MongoClient( + result_cache_config.get_uri() + ) + with results_cache_client: + results_cache_db = results_cache_client.get_default_database() + job_ids = [int(name) for name in results_cache_db.list_collection_names() if name.isdigit()] + expired_job_ids = _get_expired_job_ids(database_config, job_ids, retention_period) + for job_id in expired_job_ids: + job_id_str = str(job_id) + _delete_result_metadata(results_cache_db, results_metadata_collection_name, job_id_str) + results_cache_db.get_collection(job_id_str).drop() + deleted_job_ids.append(job_id) if len(deleted_job_ids) != 0: - logger.debug(f"Deleted search results of job(s): {deleted_job_ids}.") + logger.debug("Deleted search results of job(s): %s.", deleted_job_ids) else: logger.debug("No search results matched the expiry criteria.") async def search_result_garbage_collector(clp_config: ClpConfig) -> None: + """Run search-result collection and sweeping at the configured interval.""" configure_logging(logger, SEARCH_RESULT_GARBAGE_COLLECTOR_NAME) sweep_interval_secs = clp_config.garbage_collector.sweep_interval.search_result * MIN_TO_SECONDS - logger.info(f"{SEARCH_RESULT_GARBAGE_COLLECTOR_NAME} started.") + logger.info("%s started.", SEARCH_RESULT_GARBAGE_COLLECTOR_NAME) try: while True: _collect_and_sweep_expired_search_results( - clp_config.results_cache, clp_config.webui.results_metadata_collection_name + clp_config.results_cache, + clp_config.database, + clp_config.webui.results_metadata_collection_name, ) await asyncio.sleep(sweep_interval_secs) except Exception: - logger.exception(f"{SEARCH_RESULT_GARBAGE_COLLECTOR_NAME} exited with failure.") + logger.exception("%s exited with failure.", SEARCH_RESULT_GARBAGE_COLLECTOR_NAME) raise From 2b5deba8543a4df87172da3baaeb67c326b10900 Mon Sep 17 00:00:00 2001 From: Sitao Wang Date: Mon, 31 Aug 2026 14:30:34 -0400 Subject: [PATCH 03/27] Rename ix to idx --- components/core/src/clp_s/OutputHandlerImpl.cpp | 10 +--------- components/core/src/clp_s/archive_constants.hpp | 2 +- .../search_result_garbage_collector.py | 10 ++++------ 3 files changed, 6 insertions(+), 16 deletions(-) diff --git a/components/core/src/clp_s/OutputHandlerImpl.cpp b/components/core/src/clp_s/OutputHandlerImpl.cpp index 673d7022e2..9240d49d48 100644 --- a/components/core/src/clp_s/OutputHandlerImpl.cpp +++ b/components/core/src/clp_s/OutputHandlerImpl.cpp @@ -153,7 +153,7 @@ ErrorCode ResultsCacheOutputHandler::finish() { ), bsoncxx::builder::basic::kvp( constants::results_cache::search:: - cLogEventIx, + cLogEventIdx, result.log_event_idx ) ) @@ -170,14 +170,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) diff --git a/components/core/src/clp_s/archive_constants.hpp b/components/core/src/clp_s/archive_constants.hpp index 6dfae883e7..9d8b134842 100644 --- a/components/core/src/clp_s/archive_constants.hpp +++ b/components/core/src/clp_s/archive_constants.hpp @@ -55,7 +55,7 @@ constexpr char cIsLastChunk[]{"is_last_chunk"}; 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"}; diff --git a/components/job-orchestration/job_orchestration/garbage_collector/search_result_garbage_collector.py b/components/job-orchestration/job_orchestration/garbage_collector/search_result_garbage_collector.py index 99e574485b..f5f729ab2d 100644 --- a/components/job-orchestration/job_orchestration/garbage_collector/search_result_garbage_collector.py +++ b/components/job-orchestration/job_orchestration/garbage_collector/search_result_garbage_collector.py @@ -49,11 +49,10 @@ def _get_expired_job_ids( closing(sql_adapter.create_connection(True)) as db_conn, closing(db_conn.cursor()) as db_cursor, ): - for begin_ix in range(0, len(job_ids), MAX_NUM_JOB_IDS_PER_QUERY): - job_ids_batch = job_ids[begin_ix : begin_ix + MAX_NUM_JOB_IDS_PER_QUERY] + for begin_idx in range(0, len(job_ids), MAX_NUM_JOB_IDS_PER_QUERY): + job_ids_batch = job_ids[begin_idx : begin_idx + MAX_NUM_JOB_IDS_PER_QUERY] job_id_placeholders = ",".join(["%s"] * len(job_ids_batch)) - query = ( - f""" + query = f""" SELECT id FROM `{QUERY_JOBS_TABLE_NAME}` WHERE id IN ({job_id_placeholders}) @@ -62,8 +61,7 @@ def _get_expired_job_ids( CAST(duration * 1000000 AS SIGNED), creation_time ) < TIMESTAMPADD(MINUTE, %s, CURRENT_TIMESTAMP(3)) - """ # noqa: S608 - ) + """ db_cursor.execute( query, [*job_ids_batch, -retention_period_minutes], From d22e01a42ff5278c401c411188cde59ccbb085e5 Mon Sep 17 00:00:00 2001 From: Sitao Wang Date: Mon, 31 Aug 2026 14:31:51 -0400 Subject: [PATCH 04/27] Use _id column --- .../clp_mcp_server/clp_connector.py | 8 +++- .../tests/test_clp_connector.py | 14 +++---- .../SearchResultsVirtualTable/typings.tsx | 4 +- .../useSearchResults.ts | 39 ++++++++++++++++--- 4 files changed, 48 insertions(+), 17 deletions(-) diff --git a/components/clp-mcp-server/clp_mcp_server/clp_connector.py b/components/clp-mcp-server/clp_mcp_server/clp_connector.py index b3f158d419..544a538d14 100644 --- a/components/clp-mcp-server/clp_mcp_server/clp_connector.py +++ b/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) diff --git a/components/clp-mcp-server/tests/test_clp_connector.py b/components/clp-mcp-server/tests/test_clp_connector.py index dd5ff60115..568c45972d 100644 --- a/components/clp-mcp-server/tests/test_clp_connector.py +++ b/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 diff --git a/components/webui/packages/client/src/pages/SearchPage/SearchResults/SearchResultsTable/Native/SearchResultsVirtualTable/typings.tsx b/components/webui/packages/client/src/pages/SearchPage/SearchResults/SearchResultsTable/Native/SearchResultsVirtualTable/typings.tsx index 41852da995..f0b1177856 100644 --- a/components/webui/packages/client/src/pages/SearchPage/SearchResults/SearchResultsTable/Native/SearchResultsVirtualTable/typings.tsx +++ b/components/webui/packages/client/src/pages/SearchPage/SearchResults/SearchResultsTable/Native/SearchResultsVirtualTable/typings.tsx @@ -25,7 +25,7 @@ interface SearchResult { archive_id: string; dataset: string; filePath: string; - log_event_ix: number; + log_event_idx: number; message: string; orig_file_id: string; orig_file_path: string; @@ -90,7 +90,7 @@ const searchResultsTableColumns: NonNullable["columns"] render: (_, record) => ( & { + _id: LegacyObjectId | string; + log_event_ix: number; + }) | + (Omit & { + _id: SearchResultId; + }); + + /** * Custom hook to stream search results for the current searchJobId from the API server's SSE * endpoint. When the stream ends, the search UI state is updated to `DONE` (or `FAILED` if the @@ -33,15 +52,23 @@ const useSearchResults = () => { } }, parse: (data) => { - // MongoDB ObjectIds are serialized as `{"$oid": "..."}` in the SSE stream. - const doc = JSON.parse(data) as Omit & - {_id: string | {$oid: string}}; + const doc = JSON.parse(data) as RawSearchResult; + + if ("log_event_ix" in doc) { + return { + ...doc, + _id: "object" === typeof doc._id ? + doc._id.$oid : + doc._id, + log_event_idx: doc.log_event_ix, + }; + } return { ...doc, - _id: "object" === typeof doc._id ? - doc._id.$oid : - doc._id, + archive_id: doc._id.archive_id, + _id: JSON.stringify(doc._id), + log_event_idx: doc._id.log_event_idx, }; }, rawDocs: true, From cc8303a631e16e50144e6a523ef2f46dbdb84897 Mon Sep 17 00:00:00 2001 From: Sitao Wang Date: Mon, 31 Aug 2026 14:49:18 -0400 Subject: [PATCH 05/27] Add legacy read support --- .../clp_mcp_server/clp_connector.py | 7 +++++-- .../clp-mcp-server/tests/test_clp_connector.py | 18 ++++++++++++++++++ 2 files changed, 23 insertions(+), 2 deletions(-) diff --git a/components/clp-mcp-server/clp_mcp_server/clp_connector.py b/components/clp-mcp-server/clp_mcp_server/clp_connector.py index 544a538d14..1f0f903667 100644 --- a/components/clp-mcp-server/clp_mcp_server/clp_connector.py +++ b/components/clp-mcp-server/clp_mcp_server/clp_connector.py @@ -163,8 +163,11 @@ async def read_results(self, query_id: str) -> list[dict]: 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"] + 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']}" diff --git a/components/clp-mcp-server/tests/test_clp_connector.py b/components/clp-mcp-server/tests/test_clp_connector.py index 568c45972d..83b5b4736c 100644 --- a/components/clp-mcp-server/tests/test_clp_connector.py +++ b/components/clp-mcp-server/tests/test_clp_connector.py @@ -124,6 +124,24 @@ async def test_read_results_returns_docs(mock_clp_config: Any) -> None: 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}, + ] + 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 [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.""" From 7912f44e89e77ab17a2a8bd5776ea56f8820d788 Mon Sep 17 00:00:00 2001 From: Sitao Wang Date: Mon, 31 Aug 2026 14:50:40 -0400 Subject: [PATCH 06/27] Use start_time instead of create_time in completion time calcuation --- .../garbage_collector/search_result_garbage_collector.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/components/job-orchestration/job_orchestration/garbage_collector/search_result_garbage_collector.py b/components/job-orchestration/job_orchestration/garbage_collector/search_result_garbage_collector.py index f5f729ab2d..12a0f09778 100644 --- a/components/job-orchestration/job_orchestration/garbage_collector/search_result_garbage_collector.py +++ b/components/job-orchestration/job_orchestration/garbage_collector/search_result_garbage_collector.py @@ -1,3 +1,5 @@ +"""Garbage-collect cached search results.""" + import asyncio from contextlib import closing from typing import Any, cast, Final @@ -31,7 +33,7 @@ def _get_expired_job_ids( """ Filter query-job IDs by whether their retention periods have ended. - MariaDB computes each query's completion time as `creation_time + duration`. A query-job ID is + MariaDB computes each query's completion time as `start_time + duration`. A query-job ID is included when the time since completion is greater than `retention_period_minutes`. :param database_config: Configuration for the orchestration database. @@ -59,7 +61,7 @@ def _get_expired_job_ids( AND TIMESTAMPADD( MICROSECOND, CAST(duration * 1000000 AS SIGNED), - creation_time + start_time ) < TIMESTAMPADD(MINUTE, %s, CURRENT_TIMESTAMP(3)) """ db_cursor.execute( From 500bb2c8c22aad6a1795944bb6b0bf92cc74e943 Mon Sep 17 00:00:00 2001 From: Sitao Wang Date: Mon, 31 Aug 2026 15:10:47 -0400 Subject: [PATCH 07/27] Style improvement --- .../core/src/clp_s/OutputHandlerImpl.cpp | 150 +++++++++++++----- 1 file changed, 109 insertions(+), 41 deletions(-) diff --git a/components/core/src/clp_s/OutputHandlerImpl.cpp b/components/core/src/clp_s/OutputHandlerImpl.cpp index 9240d49d48..0249c59735 100644 --- a/components/core/src/clp_s/OutputHandlerImpl.cpp +++ b/components/core/src/clp_s/OutputHandlerImpl.cpp @@ -1,5 +1,6 @@ #include "OutputHandlerImpl.hpp" +#include #include #include #include @@ -34,6 +35,86 @@ 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(reply["code"]) || static_cast(reply["errmsg"])) { + return false; + } + + auto const command_status = reply["ok"]; + if (false == static_cast(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(reply["writeConcernError"])) { + return true; + } + + auto const errors_element = reply["writeConcernErrors"]; + if (false == static_cast(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(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 { @@ -42,36 +123,23 @@ constexpr int32_t cDuplicateKeyErrorCode{11'000}; return false; } - auto const write_errors_element = raw_server_error->view()["writeErrors"]; + 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(write_errors_element) || bsoncxx::type::k_array != write_errors_element.type()) { return false; } - bool found_write_error{false}; - for (auto const& write_error_element : write_errors_element.get_array().value) { - if (bsoncxx::type::k_document != write_error_element.type()) { - return false; - } - auto const code_element = write_error_element.get_document().value["code"]; - if (false == static_cast(code_element)) { - return false; - } - if (bsoncxx::type::k_int32 == code_element.type()) { - if (cDuplicateKeyErrorCode != code_element.get_int32().value) { - return false; - } - } else if (bsoncxx::type::k_int64 == code_element.type()) { - if (cDuplicateKeyErrorCode != code_element.get_int64().value) { - return false; - } - } else { - return false; - } - found_write_error = true; + auto const write_errors = write_errors_element.get_array().value; + if (write_errors.begin() == write_errors.end()) { + return false; } - return found_write_error; + return std::all_of(write_errors.begin(), write_errors.end(), is_duplicate_key_write_error); } } // namespace @@ -204,24 +272,6 @@ ErrorCode ResultsCacheOutputHandler::finish() { return ErrorCode::ErrorCodeSuccess; } -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; -} - void ResultsCacheOutputHandler::write( string_view message, epochtime_t timestamp, @@ -254,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), From 4d0b9fb4c6d91c34f23fb35db234879b1a5dd4b1 Mon Sep 17 00:00:00 2001 From: Sitao Wang Date: Mon, 31 Aug 2026 19:34:29 -0400 Subject: [PATCH 08/27] Address coderabbit comment --- components/clp-mcp-server/tests/test_clp_connector.py | 2 +- components/core/src/clp_s/OutputHandlerImpl.cpp | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/components/clp-mcp-server/tests/test_clp_connector.py b/components/clp-mcp-server/tests/test_clp_connector.py index 83b5b4736c..b8892c8c55 100644 --- a/components/clp-mcp-server/tests/test_clp_connector.py +++ b/components/clp-mcp-server/tests/test_clp_connector.py @@ -125,7 +125,7 @@ async def test_read_results_returns_docs(mock_clp_config: Any) -> None: @pytest.mark.asyncio -async def test_read_results_supports_legacy_docs(mock_clp_config: Any) -> None: +async def test_read_results_supports_legacy_docs(mock_clp_config: SimpleNamespace) -> None: """Tests reading legacy results with top-level archive and log-event fields.""" connector = ClpConnector(mock_clp_config) mock_docs = [ diff --git a/components/core/src/clp_s/OutputHandlerImpl.cpp b/components/core/src/clp_s/OutputHandlerImpl.cpp index 0249c59735..3119dc9648 100644 --- a/components/core/src/clp_s/OutputHandlerImpl.cpp +++ b/components/core/src/clp_s/OutputHandlerImpl.cpp @@ -47,7 +47,7 @@ constexpr int32_t cDuplicateKeyErrorCode{11'000}; auto const command_status = reply["ok"]; if (false == static_cast(command_status)) { - return true; + return false; } if (bsoncxx::type::k_double == command_status.type()) { return 1.0 == command_status.get_double().value; From 49fe7e2978cf57f3550052b44b776ddb84cb60d2 Mon Sep 17 00:00:00 2001 From: Sitao Wang Date: Mon, 31 Aug 2026 20:12:38 -0400 Subject: [PATCH 09/27] Fix exception catch --- .../core/src/clp_s/OutputHandlerImpl.cpp | 21 +++++++------------ 1 file changed, 7 insertions(+), 14 deletions(-) diff --git a/components/core/src/clp_s/OutputHandlerImpl.cpp b/components/core/src/clp_s/OutputHandlerImpl.cpp index 3119dc9648..14800de6ed 100644 --- a/components/core/src/clp_s/OutputHandlerImpl.cpp +++ b/components/core/src/clp_s/OutputHandlerImpl.cpp @@ -245,28 +245,21 @@ ErrorCode ResultsCacheOutputHandler::finish() { ) ) ); - count++; - - if (count == m_batch_size) { - 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()); + SPDLOG_ERROR("Failed to build search result - {}", e.what()); return ErrorCode::ErrorCodeFailureDbBulkWrite; } - } - try { - if (false == m_results.empty()) { + count++; + if (count == m_batch_size) { if (false == insert_results()) { return ErrorCode::ErrorCodeFailureDbBulkWrite; } + count = 0; } - } catch (mongocxx::exception const& e) { - SPDLOG_ERROR("Failed to insert final search-results batch - {}", e.what()); + } + + if (false == m_results.empty() && false == insert_results()) { return ErrorCode::ErrorCodeFailureDbBulkWrite; } return ErrorCode::ErrorCodeSuccess; From e7734fbbb210c008bbdb57c0590456cab6459508 Mon Sep 17 00:00:00 2001 From: Sitao Wang Date: Mon, 31 Aug 2026 20:31:00 -0400 Subject: [PATCH 10/27] Fix style --- .../core/src/clp_s/OutputHandlerImpl.cpp | 48 +++++++++++-------- 1 file changed, 29 insertions(+), 19 deletions(-) diff --git a/components/core/src/clp_s/OutputHandlerImpl.cpp b/components/core/src/clp_s/OutputHandlerImpl.cpp index 14800de6ed..6a4f9d6b76 100644 --- a/components/core/src/clp_s/OutputHandlerImpl.cpp +++ b/components/core/src/clp_s/OutputHandlerImpl.cpp @@ -40,6 +40,35 @@ constexpr int32_t cDuplicateKeyErrorCode{11'000}; * @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; + +/** + * 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; + +/** + * 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; + [[nodiscard]] auto is_successful_command_reply(bsoncxx::document::view const& reply) -> bool { if (static_cast(reply["code"]) || static_cast(reply["errmsg"])) { return false; @@ -61,11 +90,6 @@ constexpr int32_t cDuplicateKeyErrorCode{11'000}; 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(reply["writeConcernError"])) { return true; @@ -82,11 +106,6 @@ constexpr int32_t cDuplicateKeyErrorCode{11'000}; 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()) { @@ -106,15 +125,6 @@ constexpr int32_t cDuplicateKeyErrorCode{11'000}; 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 { From 19d477b042f4b82ee48d29ac2bbb15f999193e20 Mon Sep 17 00:00:00 2001 From: Sitao Wang Date: Mon, 31 Aug 2026 20:42:15 -0400 Subject: [PATCH 11/27] Fix error parsing --- .../core/src/clp_s/OutputHandlerImpl.cpp | 35 ++++++++----------- 1 file changed, 15 insertions(+), 20 deletions(-) diff --git a/components/core/src/clp_s/OutputHandlerImpl.cpp b/components/core/src/clp_s/OutputHandlerImpl.cpp index 6a4f9d6b76..3f13c4da6d 100644 --- a/components/core/src/clp_s/OutputHandlerImpl.cpp +++ b/components/core/src/clp_s/OutputHandlerImpl.cpp @@ -36,11 +36,11 @@ namespace { constexpr int32_t cDuplicateKeyErrorCode{11'000}; /** - * Checks whether a bulk-write reply reports a successful command. + * 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 no command error, false otherwise. + * @return true if the reply contains a command error, false otherwise. */ -[[nodiscard]] auto is_successful_command_reply(bsoncxx::document::view const& reply) -> bool; +[[nodiscard]] auto has_command_errors(bsoncxx::document::view const& reply) -> bool; /** * Checks whether a bulk-write reply contains any write-concern errors. @@ -59,6 +59,10 @@ constexpr int32_t cDuplicateKeyErrorCode{11'000}; /** * Returns whether the bulk write failed only because some documents already exist. * + * The C++ driver exposes per-write failures through `raw_server_error()`. In this aggregated reply, + * command errors are stored in `errorReplies`, write-concern errors in `writeConcernErrors`, and + * individual write failures in `writeErrors`. + * * 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. @@ -69,25 +73,16 @@ constexpr int32_t cDuplicateKeyErrorCode{11'000}; mongocxx::bulk_write_exception const& exception ) -> bool; -[[nodiscard]] auto is_successful_command_reply(bsoncxx::document::view const& reply) -> bool { - if (static_cast(reply["code"]) || static_cast(reply["errmsg"])) { - return false; - } - - auto const command_status = reply["ok"]; - if (false == static_cast(command_status)) { +[[nodiscard]] auto has_command_errors(bsoncxx::document::view const& reply) -> bool { + auto const errors_element = reply["errorReplies"]; + if (false == static_cast(errors_element)) { return false; } - 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; + if (bsoncxx::type::k_array != errors_element.type()) { + return true; } - return false; + 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 { @@ -134,7 +129,7 @@ constexpr int32_t cDuplicateKeyErrorCode{11'000}; } auto const reply = raw_server_error->view(); - if (false == is_successful_command_reply(reply) || has_write_concern_errors(reply)) { + if (has_command_errors(reply) || has_write_concern_errors(reply)) { return false; } From 3b38b902e1c2b19f0fa70cf965e762d2d4c1a36a Mon Sep 17 00:00:00 2001 From: Sitao Wang Date: Tue, 1 Sep 2026 23:02:24 -0400 Subject: [PATCH 12/27] Remove unused function and const --- .../job_orchestration/garbage_collector/utils.py | 15 --------------- 1 file changed, 15 deletions(-) diff --git a/components/job-orchestration/job_orchestration/garbage_collector/utils.py b/components/job-orchestration/job_orchestration/garbage_collector/utils.py index a53a0715a2..45aa31188d 100644 --- a/components/job-orchestration/job_orchestration/garbage_collector/utils.py +++ b/components/job-orchestration/job_orchestration/garbage_collector/utils.py @@ -1,7 +1,6 @@ import os import pathlib import shutil -import time from datetime import datetime, timezone from bson import ObjectId @@ -13,8 +12,6 @@ ) from clp_py_utils.s3_utils import s3_delete_objects -from job_orchestration.garbage_collector.constants import MIN_TO_SECONDS - def validate_storage_type(output_config: ArchiveOutput, storage_engine: str) -> None: storage_type = output_config.storage.type @@ -27,18 +24,6 @@ def validate_storage_type(output_config: ArchiveOutput, storage_engine: str) -> raise ValueError(f"Unsupported Storage type: {storage_type}") -def get_expiry_epoch_secs(retention_minutes: int) -> int: - """ - Returns a cutoff `expiry_epoch` based on the current timestamp and `retention_minutes`. Any - candidate with a timestamp (`ts`) less than `expiry_epoch` is considered expired. - The `expiry_epoch` is calculated as `expiry_epoch = cur_time - retention_secs`. - - :param retention_minutes: Retention period in minutes. - :return: The UTC epoch representing the expiry cutoff time. - """ - return int(time.time() - retention_minutes * MIN_TO_SECONDS) - - def get_oid_with_expiry_time(expiry_epoch_secs: int) -> ObjectId: return ObjectId.from_datetime(datetime.fromtimestamp(expiry_epoch_secs, tz=timezone.utc)) From 2b6d0e129b6f162d185929e1c480c3d6739d2de9 Mon Sep 17 00:00:00 2001 From: Sitao Wang Date: Wed, 2 Sep 2026 17:24:36 -0400 Subject: [PATCH 13/27] Update mongodb in clo --- components/core/src/clp/clo/CMakeLists.txt | 8 +- components/core/src/clp/clo/OutputHandler.cpp | 64 +++++++--- components/core/src/clp/clo/OutputHandler.hpp | 7 ++ components/core/src/clp/clo/constants.hpp | 3 +- components/core/src/clp_s/CMakeLists.txt | 2 + components/core/src/clp_s/MongoDBUtils.cpp | 106 ++++++++++++++++ components/core/src/clp_s/MongoDBUtils.hpp | 21 ++++ .../core/src/clp_s/OutputHandlerImpl.cpp | 119 +----------------- .../useSearchResults.ts | 21 +++- 9 files changed, 210 insertions(+), 141 deletions(-) create mode 100644 components/core/src/clp_s/MongoDBUtils.cpp create mode 100644 components/core/src/clp_s/MongoDBUtils.hpp diff --git a/components/core/src/clp/clo/CMakeLists.txt b/components/core/src/clp/clo/CMakeLists.txt index 588267dfbb..f567e2044a 100644 --- a/components/core/src/clp/clo/CMakeLists.txt +++ b/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 diff --git a/components/core/src/clp/clo/OutputHandler.cpp b/components/core/src/clp/clo/OutputHandler.cpp index 1ad61221c2..c29b41a35f 100644 --- a/components/core/src/clp/clo/OutputHandler.cpp +++ b/components/core/src/clp/clo/OutputHandler.cpp @@ -4,9 +4,13 @@ #include #include +#include +#include #include #include +#include + #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} { diff --git a/components/core/src/clp/clo/OutputHandler.hpp b/components/core/src/clp/clo/OutputHandler.hpp index b4824b5202..2c60446ff3 100644 --- a/components/core/src/clp/clo/OutputHandler.hpp +++ b/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 m_results; diff --git a/components/core/src/clp/clo/constants.hpp b/components/core/src/clp/clo/constants.hpp index 945bde83e0..295179e220 100644 --- a/components/core/src/clp/clo/constants.hpp +++ b/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 diff --git a/components/core/src/clp_s/CMakeLists.txt b/components/core/src/clp_s/CMakeLists.txt index b845622bbd..d0a625a6e9 100644 --- a/components/core/src/clp_s/CMakeLists.txt +++ b/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 diff --git a/components/core/src/clp_s/MongoDBUtils.cpp b/components/core/src/clp_s/MongoDBUtils.cpp new file mode 100644 index 0000000000..ca2daad9c0 --- /dev/null +++ b/components/core/src/clp_s/MongoDBUtils.cpp @@ -0,0 +1,106 @@ +#include "MongoDBUtils.hpp" + +#include + +#include +#include + +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(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(reply["writeConcernError"])) { + return true; + } + + auto const errors_element = reply["writeConcernErrors"]; + if (false == static_cast(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(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(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 diff --git a/components/core/src/clp_s/MongoDBUtils.hpp b/components/core/src/clp_s/MongoDBUtils.hpp new file mode 100644 index 0000000000..5546e5626d --- /dev/null +++ b/components/core/src/clp_s/MongoDBUtils.hpp @@ -0,0 +1,21 @@ +#ifndef CLP_S_MONGODBUTILS_HPP +#define CLP_S_MONGODBUTILS_HPP + +#include + +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 diff --git a/components/core/src/clp_s/OutputHandlerImpl.cpp b/components/core/src/clp_s/OutputHandlerImpl.cpp index 3f13c4da6d..a88a867470 100644 --- a/components/core/src/clp_s/OutputHandlerImpl.cpp +++ b/components/core/src/clp_s/OutputHandlerImpl.cpp @@ -1,6 +1,5 @@ #include "OutputHandlerImpl.hpp" -#include #include #include #include @@ -8,7 +7,6 @@ #include #include -#include #include #include #include @@ -18,6 +16,7 @@ #include #include +#include #include #include "../clp/networking/socket_utils.hpp" @@ -32,122 +31,6 @@ using std::string; using std::string_view; 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; - -/** - * Returns whether the bulk write failed only because some documents already exist. - * - * The C++ driver exposes per-write failures through `raw_server_error()`. In this aggregated reply, - * command errors are stored in `errorReplies`, write-concern errors in `writeConcernErrors`, and - * individual write failures in `writeErrors`. - * - * 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; - -[[nodiscard]] auto has_command_errors(bsoncxx::document::view const& reply) -> bool { - auto const errors_element = reply["errorReplies"]; - if (false == static_cast(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(reply["writeConcernError"])) { - return true; - } - - auto const errors_element = reply["writeConcernErrors"]; - if (false == static_cast(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(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; -} - -[[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 (has_command_errors(reply) || has_write_concern_errors(reply)) { - return false; - } - - auto const write_errors_element = reply["writeErrors"]; - if (false == static_cast(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, diff --git a/components/webui/packages/client/src/pages/SearchPage/SearchResults/SearchResultsTable/Native/SearchResultsVirtualTable/useSearchResults.ts b/components/webui/packages/client/src/pages/SearchPage/SearchResults/SearchResultsTable/Native/SearchResultsVirtualTable/useSearchResults.ts index 4d19eccd19..a107f28550 100644 --- a/components/webui/packages/client/src/pages/SearchPage/SearchResults/SearchResultsTable/Native/SearchResultsVirtualTable/useSearchResults.ts +++ b/components/webui/packages/client/src/pages/SearchPage/SearchResults/SearchResultsTable/Native/SearchResultsVirtualTable/useSearchResults.ts @@ -6,7 +6,12 @@ import {SEARCH_UI_STATE} from "../../../../SearchState/typings"; import {SearchResult} from "./typings"; -interface SearchResultId { +interface ClpSearchResultId { + log_event_idx: number; + orig_file_id: string; +} + +interface ClpSSearchResultId { archive_id: string; log_event_idx: number; } @@ -20,8 +25,11 @@ type RawSearchResult = _id: LegacyObjectId | string; log_event_ix: number; }) | + (Omit & { + _id: ClpSearchResultId; + }) | (Omit & { - _id: SearchResultId; + _id: ClpSSearchResultId; }); @@ -64,6 +72,15 @@ const useSearchResults = () => { }; } + if ("orig_file_id" in doc._id) { + return { + ...doc, + _id: JSON.stringify(doc._id), + log_event_idx: doc._id.log_event_idx, + orig_file_id: doc._id.orig_file_id, + }; + } + return { ...doc, archive_id: doc._id.archive_id, From b39bf3ad28a1a9ef409ede482e27c65eb108ee80 Mon Sep 17 00:00:00 2001 From: Sitao Wang Date: Wed, 2 Sep 2026 17:33:31 -0400 Subject: [PATCH 14/27] Remove backward compatibility --- .../clp_mcp_server/clp_connector.py | 7 ++----- .../clp-mcp-server/tests/test_clp_connector.py | 18 ------------------ .../useSearchResults.ts | 18 ------------------ 3 files changed, 2 insertions(+), 41 deletions(-) diff --git a/components/clp-mcp-server/clp_mcp_server/clp_connector.py b/components/clp-mcp-server/clp_mcp_server/clp_connector.py index 1f0f903667..544a538d14 100644 --- a/components/clp-mcp-server/clp_mcp_server/clp_connector.py +++ b/components/clp-mcp-server/clp_mcp_server/clp_connector.py @@ -163,11 +163,8 @@ async def read_results(self, query_id: str) -> list[dict]: 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["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']}" diff --git a/components/clp-mcp-server/tests/test_clp_connector.py b/components/clp-mcp-server/tests/test_clp_connector.py index b8892c8c55..568c45972d 100644 --- a/components/clp-mcp-server/tests/test_clp_connector.py +++ b/components/clp-mcp-server/tests/test_clp_connector.py @@ -124,24 +124,6 @@ async def test_read_results_returns_docs(mock_clp_config: Any) -> None: assert len(results) == len(mock_docs) -@pytest.mark.asyncio -async def test_read_results_supports_legacy_docs(mock_clp_config: SimpleNamespace) -> 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}, - ] - 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 [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.""" diff --git a/components/webui/packages/client/src/pages/SearchPage/SearchResults/SearchResultsTable/Native/SearchResultsVirtualTable/useSearchResults.ts b/components/webui/packages/client/src/pages/SearchPage/SearchResults/SearchResultsTable/Native/SearchResultsVirtualTable/useSearchResults.ts index a107f28550..6acae7247e 100644 --- a/components/webui/packages/client/src/pages/SearchPage/SearchResults/SearchResultsTable/Native/SearchResultsVirtualTable/useSearchResults.ts +++ b/components/webui/packages/client/src/pages/SearchPage/SearchResults/SearchResultsTable/Native/SearchResultsVirtualTable/useSearchResults.ts @@ -16,15 +16,7 @@ interface ClpSSearchResultId { log_event_idx: number; } -interface LegacyObjectId { - $oid: string; -} - type RawSearchResult = - (Omit & { - _id: LegacyObjectId | string; - log_event_ix: number; - }) | (Omit & { _id: ClpSearchResultId; }) | @@ -62,16 +54,6 @@ const useSearchResults = () => { parse: (data) => { const doc = JSON.parse(data) as RawSearchResult; - if ("log_event_ix" in doc) { - return { - ...doc, - _id: "object" === typeof doc._id ? - doc._id.$oid : - doc._id, - log_event_idx: doc.log_event_ix, - }; - } - if ("orig_file_id" in doc._id) { return { ...doc, From 07405356e93ec33bfc10fa6fd92d162ff8bb3d99 Mon Sep 17 00:00:00 2001 From: Sitao Wang Date: Wed, 2 Sep 2026 22:30:36 -0400 Subject: [PATCH 15/27] Bug fix --- .../SearchResultsVirtualTable/useSearchResults.ts | 13 ++++++++++--- 1 file changed, 10 insertions(+), 3 deletions(-) diff --git a/components/webui/packages/client/src/pages/SearchPage/SearchResults/SearchResultsTable/Native/SearchResultsVirtualTable/useSearchResults.ts b/components/webui/packages/client/src/pages/SearchPage/SearchResults/SearchResultsTable/Native/SearchResultsVirtualTable/useSearchResults.ts index 6acae7247e..2316fbf30e 100644 --- a/components/webui/packages/client/src/pages/SearchPage/SearchResults/SearchResultsTable/Native/SearchResultsVirtualTable/useSearchResults.ts +++ b/components/webui/packages/client/src/pages/SearchPage/SearchResults/SearchResultsTable/Native/SearchResultsVirtualTable/useSearchResults.ts @@ -16,11 +16,16 @@ interface ClpSSearchResultId { log_event_idx: number; } +type SearchResultWithoutId = Omit< + SearchResult, + "_id" | "archive_id" | "log_event_idx" | "orig_file_id" +>; + type RawSearchResult = - (Omit & { + (SearchResultWithoutId & { _id: ClpSearchResultId; }) | - (Omit & { + (SearchResultWithoutId & { _id: ClpSSearchResultId; }); @@ -58,6 +63,7 @@ const useSearchResults = () => { return { ...doc, _id: JSON.stringify(doc._id), + archive_id: "", log_event_idx: doc._id.log_event_idx, orig_file_id: doc._id.orig_file_id, }; @@ -65,9 +71,10 @@ const useSearchResults = () => { return { ...doc, - archive_id: doc._id.archive_id, _id: JSON.stringify(doc._id), + archive_id: doc._id.archive_id, log_event_idx: doc._id.log_event_idx, + orig_file_id: "", }; }, rawDocs: true, From 84e1074b1090ea0863544dd008b87b0f0512b08d Mon Sep 17 00:00:00 2001 From: Sitao Wang Date: Thu, 3 Sep 2026 13:22:23 -0400 Subject: [PATCH 16/27] Guard against null timestamp --- .../search_result_garbage_collector.py | 28 ++++++++++++++----- 1 file changed, 21 insertions(+), 7 deletions(-) diff --git a/components/job-orchestration/job_orchestration/garbage_collector/search_result_garbage_collector.py b/components/job-orchestration/job_orchestration/garbage_collector/search_result_garbage_collector.py index 12a0f09778..27458bce4c 100644 --- a/components/job-orchestration/job_orchestration/garbage_collector/search_result_garbage_collector.py +++ b/components/job-orchestration/job_orchestration/garbage_collector/search_result_garbage_collector.py @@ -19,6 +19,7 @@ MIN_TO_SECONDS, SEARCH_RESULT_GARBAGE_COLLECTOR_NAME, ) +from job_orchestration.scheduler.constants import QueryJobStatus # Constants MONGODB_ID_KEY: Final[str] = "_id" @@ -34,7 +35,8 @@ def _get_expired_job_ids( Filter query-job IDs by whether their retention periods have ended. MariaDB computes each query's completion time as `start_time + duration`. A query-job ID is - included when the time since completion is greater than `retention_period_minutes`. + included when the time since completion is greater than `retention_period_minutes`. For a + terminated query without a completion time, `creation_time` is used instead. :param database_config: Configuration for the orchestration database. :param job_ids: Query-job IDs to filter. @@ -58,15 +60,27 @@ def _get_expired_job_ids( SELECT id FROM `{QUERY_JOBS_TABLE_NAME}` WHERE id IN ({job_id_placeholders}) - AND TIMESTAMPADD( - MICROSECOND, - CAST(duration * 1000000 AS SIGNED), - start_time - ) < TIMESTAMPADD(MINUTE, %s, CURRENT_TIMESTAMP(3)) + AND ( + TIMESTAMPADD( + MICROSECOND, + CAST(duration * 1000000 AS SIGNED), + start_time + ) < TIMESTAMPADD(MINUTE, %s, CURRENT_TIMESTAMP(3)) + OR ( + (start_time IS NULL OR duration IS NULL) + AND status IN ( + {QueryJobStatus.SUCCEEDED}, + {QueryJobStatus.FAILED}, + {QueryJobStatus.CANCELLED}, + {QueryJobStatus.KILLED} + ) + AND creation_time < TIMESTAMPADD(MINUTE, %s, CURRENT_TIMESTAMP(3)) + ) + ) """ db_cursor.execute( query, - [*job_ids_batch, -retention_period_minutes], + [*job_ids_batch, -retention_period_minutes, -retention_period_minutes], ) rows = cast("list[tuple[int]]", db_cursor.fetchall()) expired_job_ids.extend(row[0] for row in rows) From ee8c1aa4a24d23015da9cde83d5bbf83a49f1f2b Mon Sep 17 00:00:00 2001 From: Sitao Wang Date: Thu, 3 Sep 2026 20:48:59 -0400 Subject: [PATCH 17/27] Fix docstring --- components/core/src/clp/clo/OutputHandler.hpp | 2 +- components/core/src/clp_s/MongoDBUtils.cpp | 6 +++--- components/core/src/clp_s/MongoDBUtils.hpp | 2 +- components/core/src/clp_s/OutputHandlerImpl.hpp | 2 +- 4 files changed, 6 insertions(+), 6 deletions(-) diff --git a/components/core/src/clp/clo/OutputHandler.hpp b/components/core/src/clp/clo/OutputHandler.hpp index 2c60446ff3..d87497cbc5 100644 --- a/components/core/src/clp/clo/OutputHandler.hpp +++ b/components/core/src/clp/clo/OutputHandler.hpp @@ -211,7 +211,7 @@ class ResultsCacheOutputHandler : public OutputHandler { /** * 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. + * @return Whether insertion succeeded or produced only duplicate-key errors. */ [[nodiscard]] auto insert_results() -> bool; diff --git a/components/core/src/clp_s/MongoDBUtils.cpp b/components/core/src/clp_s/MongoDBUtils.cpp index ca2daad9c0..ca46b2e336 100644 --- a/components/core/src/clp_s/MongoDBUtils.cpp +++ b/components/core/src/clp_s/MongoDBUtils.cpp @@ -12,21 +12,21 @@ 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. + * @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 true if the reply contains a write-concern error, false otherwise. + * @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 true if the entry has MongoDB's duplicate-key error code, false otherwise. + * @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; diff --git a/components/core/src/clp_s/MongoDBUtils.hpp b/components/core/src/clp_s/MongoDBUtils.hpp index 5546e5626d..7f6a0c2916 100644 --- a/components/core/src/clp_s/MongoDBUtils.hpp +++ b/components/core/src/clp_s/MongoDBUtils.hpp @@ -11,7 +11,7 @@ namespace clp_s { * 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. + * @return Whether the reply contains only duplicate-key write errors. */ [[nodiscard]] auto contains_only_duplicate_key_write_errors( mongocxx::bulk_write_exception const& exception diff --git a/components/core/src/clp_s/OutputHandlerImpl.hpp b/components/core/src/clp_s/OutputHandlerImpl.hpp index cdaa24eee3..c9bfb18a5c 100644 --- a/components/core/src/clp_s/OutputHandlerImpl.hpp +++ b/components/core/src/clp_s/OutputHandlerImpl.hpp @@ -203,7 +203,7 @@ class ResultsCacheOutputHandler : public ::clp_s::search::OutputHandler { /** * 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. + * @return Whether insertion succeeded or produced only duplicate-key errors. */ [[nodiscard]] auto insert_results() -> bool; From 6f214b49f1da999e9dc21985421339e0f708106b Mon Sep 17 00:00:00 2001 From: Sitao Wang Date: Thu, 3 Sep 2026 20:51:03 -0400 Subject: [PATCH 18/27] Remove unnecessary copy --- components/clp-mcp-server/clp_mcp_server/clp_connector.py | 3 +-- components/clp-mcp-server/tests/test_clp_connector.py | 4 ++-- 2 files changed, 3 insertions(+), 4 deletions(-) diff --git a/components/clp-mcp-server/clp_mcp_server/clp_connector.py b/components/clp-mcp-server/clp_mcp_server/clp_connector.py index 544a538d14..8d95885121 100644 --- a/components/clp-mcp-server/clp_mcp_server/clp_connector.py +++ b/components/clp-mcp-server/clp_mcp_server/clp_connector.py @@ -160,8 +160,7 @@ async def read_results(self, query_id: str) -> list[dict]: collection = self._results_cache[str(query_id)] results = [] - async for raw_doc in collection.find({}, limit=SEARCH_MAX_NUM_RESULTS): - doc = dict(raw_doc) + 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"] diff --git a/components/clp-mcp-server/tests/test_clp_connector.py b/components/clp-mcp-server/tests/test_clp_connector.py index 568c45972d..59b31c6aaa 100644 --- a/components/clp-mcp-server/tests/test_clp_connector.py +++ b/components/clp-mcp-server/tests/test_clp_connector.py @@ -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['_id']['archive_id']}" - f"&dataset=default&logEventIdx={original['_id']['log_event_idx']}" + f"/streamFile?type=json&streamId={original['archive_id']}" + f"&dataset=default&logEventIdx={original['log_event_idx']}" ) assert result["link"] == expected_link From b450ab9203854378289f43312834986f24ca7c9f Mon Sep 17 00:00:00 2001 From: Sitao Wang Date: Thu, 3 Sep 2026 20:58:22 -0400 Subject: [PATCH 19/27] Restore f-string fix --- .../garbage_collector/search_result_garbage_collector.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/components/job-orchestration/job_orchestration/garbage_collector/search_result_garbage_collector.py b/components/job-orchestration/job_orchestration/garbage_collector/search_result_garbage_collector.py index 27458bce4c..af70ae1c5d 100644 --- a/components/job-orchestration/job_orchestration/garbage_collector/search_result_garbage_collector.py +++ b/components/job-orchestration/job_orchestration/garbage_collector/search_result_garbage_collector.py @@ -128,7 +128,7 @@ def _collect_and_sweep_expired_search_results( deleted_job_ids.append(job_id) if len(deleted_job_ids) != 0: - logger.debug("Deleted search results of job(s): %s.", deleted_job_ids) + logger.debug(f"Deleted search results of job(s): {deleted_job_ids}.") else: logger.debug("No search results matched the expiry criteria.") @@ -139,7 +139,7 @@ async def search_result_garbage_collector(clp_config: ClpConfig) -> None: sweep_interval_secs = clp_config.garbage_collector.sweep_interval.search_result * MIN_TO_SECONDS - logger.info("%s started.", SEARCH_RESULT_GARBAGE_COLLECTOR_NAME) + logger.info(f"{SEARCH_RESULT_GARBAGE_COLLECTOR_NAME} started.") try: while True: _collect_and_sweep_expired_search_results( @@ -149,5 +149,5 @@ async def search_result_garbage_collector(clp_config: ClpConfig) -> None: ) await asyncio.sleep(sweep_interval_secs) except Exception: - logger.exception("%s exited with failure.", SEARCH_RESULT_GARBAGE_COLLECTOR_NAME) + logger.exception(f"{SEARCH_RESULT_GARBAGE_COLLECTOR_NAME} exited with failure.") raise From 09ad0bc270643f34434673b389c37f3751293bb7 Mon Sep 17 00:00:00 2001 From: Sitao Wang Date: Thu, 3 Sep 2026 21:32:18 -0400 Subject: [PATCH 20/27] Fix line break --- components/core/src/clp_s/MongoDBUtils.hpp | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/components/core/src/clp_s/MongoDBUtils.hpp b/components/core/src/clp_s/MongoDBUtils.hpp index 7f6a0c2916..1b273541b9 100644 --- a/components/core/src/clp_s/MongoDBUtils.hpp +++ b/components/core/src/clp_s/MongoDBUtils.hpp @@ -7,9 +7,9 @@ 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. + * 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 Whether the reply contains only duplicate-key write errors. */ From ee064c85935f62e4de81badf09f7b78caeb461fc Mon Sep 17 00:00:00 2001 From: Sitao Wang Date: Thu, 3 Sep 2026 21:40:46 -0400 Subject: [PATCH 21/27] Use empty instead of iterator compare --- components/core/src/clp_s/MongoDBUtils.cpp | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/components/core/src/clp_s/MongoDBUtils.cpp b/components/core/src/clp_s/MongoDBUtils.cpp index ca46b2e336..50f449fd1d 100644 --- a/components/core/src/clp_s/MongoDBUtils.cpp +++ b/components/core/src/clp_s/MongoDBUtils.cpp @@ -39,7 +39,7 @@ constexpr int32_t cDuplicateKeyErrorCode{11'000}; return true; } auto const errors = errors_element.get_array().value; - return errors.begin() != errors.end(); + return false == errors.empty(); } [[nodiscard]] auto has_write_concern_errors(bsoncxx::document::view const& reply) -> bool { @@ -55,7 +55,7 @@ constexpr int32_t cDuplicateKeyErrorCode{11'000}; return true; } auto const errors = errors_element.get_array().value; - return errors.begin() != errors.end(); + return false == errors.empty(); } [[nodiscard]] auto is_duplicate_key_write_error(bsoncxx::array::element const& write_error) @@ -98,7 +98,7 @@ auto contains_only_duplicate_key_write_errors(mongocxx::bulk_write_exception con } auto const write_errors = write_errors_element.get_array().value; - if (write_errors.begin() == write_errors.end()) { + if (write_errors.empty()) { return false; } return std::all_of(write_errors.begin(), write_errors.end(), is_duplicate_key_write_error); From fd7e1078d68b473d95f7cf7735a07ab100d7f29e Mon Sep 17 00:00:00 2001 From: Sitao Wang Date: Thu, 3 Sep 2026 22:00:09 -0400 Subject: [PATCH 22/27] Make option reusable --- components/core/src/clp/clo/OutputHandler.cpp | 6 ++---- components/core/src/clp/clo/OutputHandler.hpp | 2 ++ components/core/src/clp_s/OutputHandlerImpl.cpp | 6 ++---- components/core/src/clp_s/OutputHandlerImpl.hpp | 2 ++ 4 files changed, 8 insertions(+), 8 deletions(-) diff --git a/components/core/src/clp/clo/OutputHandler.cpp b/components/core/src/clp/clo/OutputHandler.cpp index c29b41a35f..44d0a1e52b 100644 --- a/components/core/src/clp/clo/OutputHandler.cpp +++ b/components/core/src/clp/clo/OutputHandler.cpp @@ -5,7 +5,6 @@ #include #include -#include #include #include @@ -55,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); @@ -161,9 +161,7 @@ ErrorCode ResultsCacheOutputHandler::flush() { auto ResultsCacheOutputHandler::insert_results() -> bool { try { - mongocxx::options::insert options; - options.ordered(false); - m_collection.insert_many(m_results, options); + 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)) { SPDLOG_ERROR("Failed to insert search results - {}", exception.what()); diff --git a/components/core/src/clp/clo/OutputHandler.hpp b/components/core/src/clp/clo/OutputHandler.hpp index d87497cbc5..7611efd086 100644 --- a/components/core/src/clp/clo/OutputHandler.hpp +++ b/components/core/src/clp/clo/OutputHandler.hpp @@ -10,6 +10,7 @@ #include #include #include +#include #include #include "../../reducer/Pipeline.hpp" @@ -217,6 +218,7 @@ class ResultsCacheOutputHandler : public OutputHandler { mongocxx::client m_client; mongocxx::collection m_collection; + mongocxx::options::insert m_insert_options; std::vector m_results; uint64_t m_batch_size; uint64_t m_max_num_results; diff --git a/components/core/src/clp_s/OutputHandlerImpl.cpp b/components/core/src/clp_s/OutputHandlerImpl.cpp index a88a867470..f356623975 100644 --- a/components/core/src/clp_s/OutputHandlerImpl.cpp +++ b/components/core/src/clp_s/OutputHandlerImpl.cpp @@ -12,7 +12,6 @@ #include #include #include -#include #include #include @@ -86,6 +85,7 @@ ResultsCacheOutputHandler::ResultsCacheOutputHandler( m_max_num_results{max_num_results}, m_dataset{dataset} { m_collection = connect_to_results_cache(uri, collection, m_client); + m_insert_options.ordered(false); m_results.reserve(m_batch_size); } @@ -187,9 +187,7 @@ void ResultsCacheOutputHandler::write( auto ResultsCacheOutputHandler::insert_results() -> bool { try { - mongocxx::options::insert options; - options.ordered(false); - m_collection.insert_many(m_results, options); + m_collection.insert_many(m_results, m_insert_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()); diff --git a/components/core/src/clp_s/OutputHandlerImpl.hpp b/components/core/src/clp_s/OutputHandlerImpl.hpp index c9bfb18a5c..8afb3a1587 100644 --- a/components/core/src/clp_s/OutputHandlerImpl.hpp +++ b/components/core/src/clp_s/OutputHandlerImpl.hpp @@ -18,6 +18,7 @@ #include #include +#include #include #include @@ -209,6 +210,7 @@ class ResultsCacheOutputHandler : public ::clp_s::search::OutputHandler { mongocxx::client m_client; mongocxx::collection m_collection; + mongocxx::options::insert m_insert_options; std::vector m_results; uint64_t m_batch_size; uint64_t m_max_num_results; From 35b8a103ddaee49f9fa2c3798aec81e1ec76c15b Mon Sep 17 00:00:00 2001 From: Sitao Wang Date: Thu, 3 Sep 2026 22:06:19 -0400 Subject: [PATCH 23/27] Fix clang-tidy --- components/core/src/clp_s/MongoDBUtils.cpp | 1 + 1 file changed, 1 insertion(+) diff --git a/components/core/src/clp_s/MongoDBUtils.cpp b/components/core/src/clp_s/MongoDBUtils.cpp index 50f449fd1d..ed33ce6cac 100644 --- a/components/core/src/clp_s/MongoDBUtils.cpp +++ b/components/core/src/clp_s/MongoDBUtils.cpp @@ -1,6 +1,7 @@ #include "MongoDBUtils.hpp" #include +#include #include #include From 34342c60ce963ec9d0cfee0135a125d3ae78d2f1 Mon Sep 17 00:00:00 2001 From: Sitao Wang Date: Thu, 3 Sep 2026 23:07:24 -0400 Subject: [PATCH 24/27] Fix move --- components/core/src/clp_s/OutputHandlerImpl.cpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/components/core/src/clp_s/OutputHandlerImpl.cpp b/components/core/src/clp_s/OutputHandlerImpl.cpp index f356623975..0e67060367 100644 --- a/components/core/src/clp_s/OutputHandlerImpl.cpp +++ b/components/core/src/clp_s/OutputHandlerImpl.cpp @@ -105,7 +105,7 @@ ErrorCode ResultsCacheOutputHandler::finish() { bsoncxx::builder::basic::kvp( constants::results_cache::search:: cArchiveId, - result.archive_id + std::move(result.archive_id) ), bsoncxx::builder::basic::kvp( constants::results_cache::search:: From bcad695b680e61d848b25c1d27bcb9d13a795764 Mon Sep 17 00:00:00 2001 From: Sitao Wang Date: Thu, 3 Sep 2026 23:20:15 -0400 Subject: [PATCH 25/27] Fix docstring --- .../search_result_garbage_collector.py | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/components/job-orchestration/job_orchestration/garbage_collector/search_result_garbage_collector.py b/components/job-orchestration/job_orchestration/garbage_collector/search_result_garbage_collector.py index af70ae1c5d..9e4689f6f6 100644 --- a/components/job-orchestration/job_orchestration/garbage_collector/search_result_garbage_collector.py +++ b/components/job-orchestration/job_orchestration/garbage_collector/search_result_garbage_collector.py @@ -32,17 +32,17 @@ def _get_expired_job_ids( database_config: Database, job_ids: list[int], retention_period_minutes: int ) -> list[int]: """ - Filter query-job IDs by whether their retention periods have ended. + Filters query job IDs by whether their retention periods have ended. - MariaDB computes each query's completion time as `start_time + duration`. A query-job ID is + MariaDB computes each query's completion time as `start_time + duration`. A query job ID is included when the time since completion is greater than `retention_period_minutes`. For a terminated query without a completion time, `creation_time` is used instead. :param database_config: Configuration for the orchestration database. - :param job_ids: Query-job IDs to filter. + :param job_ids: Query job IDs to filter. :param retention_period_minutes: Length of the retention period following query completion, in minutes. - :return: Query-job IDs completed more than `retention_period_minutes` ago. + :return: Query job IDs completed more than `retention_period_minutes` ago. """ if len(job_ids) == 0: return [] @@ -100,9 +100,9 @@ def _collect_and_sweep_expired_search_results( results_metadata_collection_name: str, ) -> None: """ - Remove search results whose query completion time is older than the retention cutoff. + Removes search results whose query completion time is older than the retention cutoff. - Numeric MongoDB collection names are interpreted as query-job IDs. Collections selected by + Numeric MongoDB collection names are interpreted as query job IDs. Collections selected by `_get_expired_job_ids` are dropped along with their result metadata documents. :param result_cache_config: MongoDB result-cache and retention configuration. From 084c94f51f293a41c44cfa451b72b116c12f87ce Mon Sep 17 00:00:00 2001 From: Sitao Wang Date: Fri, 4 Sep 2026 19:56:28 -0400 Subject: [PATCH 26/27] Fix transport error --- components/core/src/clp/clo/OutputHandler.cpp | 2 +- components/core/src/clp_s/MongoDBUtils.cpp | 29 ++++++++++++++++--- components/core/src/clp_s/MongoDBUtils.hpp | 13 ++++++--- .../core/src/clp_s/OutputHandlerImpl.cpp | 2 +- 4 files changed, 36 insertions(+), 10 deletions(-) diff --git a/components/core/src/clp/clo/OutputHandler.cpp b/components/core/src/clp/clo/OutputHandler.cpp index 44d0a1e52b..c56d57b92a 100644 --- a/components/core/src/clp/clo/OutputHandler.cpp +++ b/components/core/src/clp/clo/OutputHandler.cpp @@ -163,7 +163,7 @@ 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)) { + 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; } diff --git a/components/core/src/clp_s/MongoDBUtils.cpp b/components/core/src/clp_s/MongoDBUtils.cpp index ed33ce6cac..990771d62f 100644 --- a/components/core/src/clp_s/MongoDBUtils.cpp +++ b/components/core/src/clp_s/MongoDBUtils.cpp @@ -1,6 +1,5 @@ #include "MongoDBUtils.hpp" -#include #include #include @@ -79,8 +78,10 @@ constexpr int32_t cDuplicateKeyErrorCode{11'000}; } } // namespace -auto contains_only_duplicate_key_write_errors(mongocxx::bulk_write_exception const& exception) - -> bool { +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; @@ -102,6 +103,26 @@ auto contains_only_duplicate_key_write_errors(mongocxx::bulk_write_exception con if (write_errors.empty()) { return false; } - return std::all_of(write_errors.begin(), write_errors.end(), is_duplicate_key_write_error); + + 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(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(num_inserted) + num_write_errors == num_documents; } } // namespace clp_s diff --git a/components/core/src/clp_s/MongoDBUtils.hpp b/components/core/src/clp_s/MongoDBUtils.hpp index 1b273541b9..993d31ca81 100644 --- a/components/core/src/clp_s/MongoDBUtils.hpp +++ b/components/core/src/clp_s/MongoDBUtils.hpp @@ -1,6 +1,8 @@ #ifndef CLP_S_MONGODBUTILS_HPP #define CLP_S_MONGODBUTILS_HPP +#include + #include namespace clp_s { @@ -8,13 +10,16 @@ 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. + * 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. - * @return Whether the reply contains only duplicate-key write errors. + * @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 + mongocxx::bulk_write_exception const& exception, + size_t num_documents ) -> bool; } // namespace clp_s diff --git a/components/core/src/clp_s/OutputHandlerImpl.cpp b/components/core/src/clp_s/OutputHandlerImpl.cpp index 0e67060367..5730ce84b2 100644 --- a/components/core/src/clp_s/OutputHandlerImpl.cpp +++ b/components/core/src/clp_s/OutputHandlerImpl.cpp @@ -189,7 +189,7 @@ auto ResultsCacheOutputHandler::insert_results() -> bool { try { m_collection.insert_many(m_results, m_insert_options); } catch (mongocxx::bulk_write_exception const& exception) { - if (false == contains_only_duplicate_key_write_errors(exception)) { + if (false == contains_only_duplicate_key_write_errors(exception, m_results.size())) { SPDLOG_ERROR("Failed to insert search results - {}", exception.what()); return false; } From bb65d006f33693c78456ec612756fbf63b0658f5 Mon Sep 17 00:00:00 2001 From: sitaowang1998 Date: Tue, 8 Sep 2026 21:52:39 -0400 Subject: [PATCH 27/27] Apply suggestions from code review Co-authored-by: Lin Zhihao <59785146+LinZhihao-723@users.noreply.github.com> --- .../garbage_collector/search_result_garbage_collector.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/components/job-orchestration/job_orchestration/garbage_collector/search_result_garbage_collector.py b/components/job-orchestration/job_orchestration/garbage_collector/search_result_garbage_collector.py index 9e4689f6f6..8f5683d009 100644 --- a/components/job-orchestration/job_orchestration/garbage_collector/search_result_garbage_collector.py +++ b/components/job-orchestration/job_orchestration/garbage_collector/search_result_garbage_collector.py @@ -134,7 +134,7 @@ def _collect_and_sweep_expired_search_results( async def search_result_garbage_collector(clp_config: ClpConfig) -> None: - """Run search-result collection and sweeping at the configured interval.""" + """Runs search-result collection and sweeping at the configured interval.""" configure_logging(logger, SEARCH_RESULT_GARBAGE_COLLECTOR_NAME) sweep_interval_secs = clp_config.garbage_collector.sweep_interval.search_result * MIN_TO_SECONDS