Skip to content
Merged
Show file tree
Hide file tree
Changes from 6 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions components/core/CMakeLists.txt
Original file line number Diff line number Diff line change
Expand Up @@ -633,6 +633,7 @@ set(SOURCE_FILES_unitTest
tests/test-BoundedReader.cpp
tests/test-BufferedFileReader.cpp
tests/test-clp_s-end_to_end.cpp
tests/test-clp_s-range_index.cpp
Comment thread
gibber9809 marked this conversation as resolved.
Resolved
tests/test-clp_s-search.cpp
tests/test-EncodedVariableInterpreter.cpp
tests/test-encoding_methods.cpp
Expand Down
4 changes: 4 additions & 0 deletions components/core/src/clp_s/ArchiveReader.hpp
Original file line number Diff line number Diff line change
Expand Up @@ -117,6 +117,10 @@ class ArchiveReader {

std::shared_ptr<ReaderUtils::SchemaMap> get_schema_map() { return m_schema_map; }

auto get_range_index() const -> std::vector<ArchiveReaderAdaptor::RangeIndexEntry> const& {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think this method will be used when we have search support?

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, that is the plan

return m_archive_reader_adaptor->get_range_index();
}

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🧹 Nitpick (assertive)

Expose intent & safety: mark get_range_index() noexcept / [[nodiscard]]

The getter already returns a const&, avoiding a copy.
Adding noexcept documents that it never throws, and [[nodiscard]] warns callers who silently ignore the returned data.

-    auto get_range_index() const -> std::vector<ArchiveReaderAdaptor::RangeIndexEntry> const& {
+    [[nodiscard]] auto get_range_index() const noexcept
+            -> std::vector<ArchiveReaderAdaptor::RangeIndexEntry> const&
+    {
         return m_archive_reader_adaptor->get_range_index();
     }
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
auto get_range_index() const -> std::vector<ArchiveReaderAdaptor::RangeIndexEntry> const& {
return m_archive_reader_adaptor->get_range_index();
}
[[nodiscard]] auto get_range_index() const noexcept
-> std::vector<ArchiveReaderAdaptor::RangeIndexEntry> const&
{
return m_archive_reader_adaptor->get_range_index();
}

/**
* Writes decoded messages to a file.
* @param writer
Expand Down
54 changes: 54 additions & 0 deletions components/core/src/clp_s/ArchiveReaderAdaptor.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -6,15 +6,18 @@
#include <optional>
#include <string>
#include <string_view>
#include <utility>
#include <vector>

#include <msgpack.hpp>
#include <nlohmann/json.hpp>
#include <spdlog/spdlog.h>

#include "../clp/BoundedReader.hpp"
#include "../clp/FileReader.hpp"
#include "archive_constants.hpp"
#include "InputConfig.hpp"
#include "RangeIndexWriter.hpp"
#include "ReaderUtils.hpp"
#include "SingleFileArchiveDefs.hpp"

Expand Down Expand Up @@ -97,6 +100,54 @@ ErrorCode ArchiveReaderAdaptor::try_read_archive_info(ZstdDecompressor& decompre
return ErrorCodeSuccess;
}

auto ArchiveReaderAdaptor::try_read_range_index(ZstdDecompressor& decompressor, size_t size)
-> ErrorCode {
std::vector<char> buffer(size);
if (auto const rc = decompressor.try_read_exact_length(buffer.data(), buffer.size());
ErrorCodeSuccess != rc)
{
return rc;
}

auto range_index_json = nlohmann::json::from_msgpack(buffer.begin(), buffer.end(), true, false);
if (false == range_index_json.is_array()) {
return ErrorCodeCorrupt;
}
Comment on lines +112 to +115

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🛠️ Refactor suggestion

Check for JSON parsing errors.

The from_msgpack call has throw_on_error set to false, but there's no explicit check if the JSON was successfully parsed before proceeding to validate it as an array. This could lead to working with invalid data.

-    auto range_index_json = nlohmann::json::from_msgpack(buffer.begin(), buffer.end(), true, false);
-    if (false == range_index_json.is_array()) {
+    auto range_index_json = nlohmann::json::from_msgpack(buffer.begin(), buffer.end(), true, false);
+    if (range_index_json.is_discarded() || false == range_index_json.is_array()) {
        return ErrorCodeCorrupt;
    }
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
auto range_index_json = nlohmann::json::from_msgpack(buffer.begin(), buffer.end(), true, false);
if (false == range_index_json.is_array()) {
return ErrorCodeCorrupt;
}
auto range_index_json = nlohmann::json::from_msgpack(buffer.begin(), buffer.end(), true, false);
if (range_index_json.is_discarded() || false == range_index_json.is_array()) {
return ErrorCodeCorrupt;
}


for (auto& range_index_entry : range_index_json) {
if (false == range_index_entry.contains(RangeIndexWriter::cStartIndexName)
|| false == range_index_entry.at(RangeIndexWriter::cStartIndexName).is_number_integer())
{
return ErrorCodeCorrupt;
}
if (false == range_index_entry.contains(RangeIndexWriter::cEndIndexName)
|| false == range_index_entry.at(RangeIndexWriter::cEndIndexName).is_number_integer())
{
return ErrorCodeCorrupt;
}
if (false == range_index_entry.contains(RangeIndexWriter::cMetadataFieldsName)
|| false == range_index_entry.at(RangeIndexWriter::cMetadataFieldsName).is_object())
{
return ErrorCodeCorrupt;
}
auto start_index{
range_index_entry.at(RangeIndexWriter::cStartIndexName).template get<size_t>()
};
auto end_index{
range_index_entry.at(RangeIndexWriter::cEndIndexName).template get<size_t>()
};
if (start_index > end_index) {
return ErrorCodeCorrupt;
}
m_range_index.emplace_back(
start_index,
end_index,
std::move(range_index_entry.at(RangeIndexWriter::cMetadataFieldsName))
);
}
Comment thread
gibber9809 marked this conversation as resolved.
Resolved
return ErrorCodeSuccess;
}
Comment on lines +102 to +153

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🛠️ Refactor suggestion

Harden range-index parsing against oversized or empty packets

A malicious or corrupt archive could set packet_size to an unreasonably large value, leading to a huge std::vector<char> allocation and potential OOM/DoS.
Likewise, a size of 0 is currently handed to nlohmann::json::from_msgpack, which will throw.

Consider:

  1. Rejecting sizes outside a sane upper bound (e.g., 16 MiB – adjust to your threat model).
  2. Pre-reserving the m_range_index vector to avoid re-allocations once the JSON array size is known.
@@
-    std::vector<char> buffer(size);
+    constexpr size_t kMaxRangeIndexPacketSize = 16 * 1024 * 1024;  // 16 MiB
+    if (0 == size || size > kMaxRangeIndexPacketSize) {
+        return ErrorCodeCorrupt;
+    }
+    std::vector<char> buffer(size);
@@
-    auto range_index_json = nlohmann::json::from_msgpack(buffer.begin(), buffer.end(), true, false);
+    auto range_index_json =
+            nlohmann::json::from_msgpack(buffer.begin(), buffer.end(), /*convert=*/true,
+                                         /*throw_on_error=*/false);
@@
-    for (auto& range_index_entry : range_index_json) {
+    m_range_index.reserve(range_index_json.size());
+    for (auto& range_index_entry : range_index_json) {

This keeps memory usage predictable and prevents undefined behaviour on zero-length payloads.

📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
auto ArchiveReaderAdaptor::try_read_range_index(ZstdDecompressor& decompressor, size_t size)
-> ErrorCode {
std::vector<char> buffer(size);
if (auto const rc = decompressor.try_read_exact_length(buffer.data(), buffer.size());
ErrorCodeSuccess != rc)
{
return rc;
}
auto range_index_json = nlohmann::json::from_msgpack(buffer.begin(), buffer.end(), true, false);
if (false == range_index_json.is_array()) {
return ErrorCodeCorrupt;
}
for (auto& range_index_entry : range_index_json) {
if (false == range_index_entry.contains(RangeIndexWriter::cStartIndexName)
|| false == range_index_entry.at(RangeIndexWriter::cStartIndexName).is_number_integer())
{
return ErrorCodeCorrupt;
}
if (false == range_index_entry.contains(RangeIndexWriter::cEndIndexName)
|| false == range_index_entry.at(RangeIndexWriter::cEndIndexName).is_number_integer())
{
return ErrorCodeCorrupt;
}
if (false == range_index_entry.contains(RangeIndexWriter::cMetadataFieldsName)
|| false == range_index_entry.at(RangeIndexWriter::cMetadataFieldsName).is_object())
{
return ErrorCodeCorrupt;
}
auto start_index{
range_index_entry.at(RangeIndexWriter::cStartIndexName).template get<size_t>()
};
auto end_index{
range_index_entry.at(RangeIndexWriter::cEndIndexName).template get<size_t>()
};
if (start_index > end_index) {
return ErrorCodeCorrupt;
}
m_range_index.emplace_back(
start_index,
end_index,
std::move(range_index_entry.at(RangeIndexWriter::cMetadataFieldsName))
);
}
return ErrorCodeSuccess;
}
auto ArchiveReaderAdaptor::try_read_range_index(ZstdDecompressor& decompressor, size_t size)
-> ErrorCode {
// Reject empty or overly large packets to prevent OOM/DoS
constexpr size_t kMaxRangeIndexPacketSize = 16 * 1024 * 1024; // 16 MiB
if (0 == size || size > kMaxRangeIndexPacketSize) {
return ErrorCodeCorrupt;
}
std::vector<char> buffer(size);
if (auto const rc = decompressor.try_read_exact_length(buffer.data(), buffer.size());
ErrorCodeSuccess != rc)
{
return rc;
}
auto range_index_json =
nlohmann::json::from_msgpack(buffer.begin(), buffer.end(),
/*convert=*/true,
/*throw_on_error=*/false);
if (false == range_index_json.is_array()) {
return ErrorCodeCorrupt;
}
// Reserve up front once we know the JSON array size
m_range_index.reserve(range_index_json.size());
for (auto& range_index_entry : range_index_json) {
if (false == range_index_entry.contains(RangeIndexWriter::cStartIndexName)
|| false == range_index_entry.at(RangeIndexWriter::cStartIndexName).is_number_integer())
{
return ErrorCodeCorrupt;
}
if (false == range_index_entry.contains(RangeIndexWriter::cEndIndexName)
|| false == range_index_entry.at(RangeIndexWriter::cEndIndexName).is_number_integer())
{
return ErrorCodeCorrupt;
}
if (false == range_index_entry.contains(RangeIndexWriter::cMetadataFieldsName)
|| false == range_index_entry.at(RangeIndexWriter::cMetadataFieldsName).is_object())
{
return ErrorCodeCorrupt;
}
auto start_index{
range_index_entry.at(RangeIndexWriter::cStartIndexName).template get<size_t>()
};
auto end_index{
range_index_entry.at(RangeIndexWriter::cEndIndexName).template get<size_t>()
};
if (start_index > end_index) {
return ErrorCodeCorrupt;
}
m_range_index.emplace_back(
start_index,
end_index,
std::move(range_index_entry.at(RangeIndexWriter::cMetadataFieldsName))
);
}
return ErrorCodeSuccess;
}


auto
ArchiveReaderAdaptor::try_read_unknown_metadata_packet(ZstdDecompressor& decompressor, size_t size)
-> ErrorCode {
Expand Down Expand Up @@ -181,6 +232,9 @@ ErrorCode ArchiveReaderAdaptor::try_read_archive_metadata(ZstdDecompressor& deco
case ArchiveMetadataPacketType::ArchiveInfo:
rc = try_read_archive_info(decompressor, packet_size);
break;
case ArchiveMetadataPacketType::RangeIndex:
rc = try_read_range_index(decompressor, packet_size);
break;
default:
rc = try_read_unknown_metadata_packet(decompressor, packet_size);
break;
Expand Down
33 changes: 33 additions & 0 deletions components/core/src/clp_s/ArchiveReaderAdaptor.hpp
Original file line number Diff line number Diff line change
@@ -1,10 +1,20 @@
#ifndef CLP_S_ARCHIVEREADERADAPTOR_HPP
#define CLP_S_ARCHIVEREADERADAPTOR_HPP

#include <cstddef>
#include <memory>
#include <optional>
#include <string>
#include <string_view>
#include <utility>
#include <vector>

// We use NOLINTNEXTLINE to satisfy clang-tidy here because while we don't use any symbols from
// `nlohmann/json.hpp` directly this code does not compile without the definition of
// `nlohmann::basic_json<>` found in the `nlohmann/json.hpp` header.
// NOLINTNEXTLINE(misc-include-cleaner)
#include <nlohmann/json.hpp>
#include <nlohmann/json_fwd.hpp>
Comment thread
gibber9809 marked this conversation as resolved.
Resolved

#include "../clp/BoundedReader.hpp"
#include "../clp/ReaderInterface.hpp"
Expand All @@ -28,6 +38,18 @@ class ArchiveReaderAdaptor {
: TraceableException(error_code, filename, line_number) {}
};

struct RangeIndexEntry {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Not sure if we should put it inside ArchiveReaderAdopter

@gibber9809 gibber9809 May 14, 2025 •

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Would you prefer moving it outside of ArchiveReaderAdaptor and still defining it in this file, or moving it to a separate header?

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Going to assume that moving it out of ArchiveReaderAdaptor but keeping it in the same file is fine for now.

@wraymo wraymo May 15, 2025 •

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I prefer moving it outside ArchiveReaderAdaptor.h. And since it's very similar to Range in RangeIndexWriter, is it possible to merge them into one struct?

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Merging them might be a bit awkward because in RangeIndexWriter we need to know whether end_index has been specified so we wrap that field in optional but on the read-side we know that it always exists so wrapping it in optional would be cumbersome.

explicit RangeIndexEntry(size_t start_index, size_t end_index, nlohmann::json&& fields)
: start_index{start_index},
end_index{end_index},
// Note: brace initializer would make nlohmann wrap the fields object in an array.
fields(std::move(fields)) {}

size_t start_index;
size_t end_index;
nlohmann::json fields;
};

explicit ArchiveReaderAdaptor(Path const& archive_path, NetworkAuthOption const& network_auth);

/**
Expand Down Expand Up @@ -62,6 +84,8 @@ class ArchiveReaderAdaptor {

ArchiveHeader const& get_header() const { return m_archive_header; }

std::vector<RangeIndexEntry> const& get_range_index() const { return m_range_index; }

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🧹 Nitpick (assertive)

Consider adding [[nodiscard]] and noexcept to get_range_index().

For consistency with other getters and to provide better guarantees to callers:

-    std::vector<RangeIndexEntry> const& get_range_index() const { return m_range_index; }
+    [[nodiscard]] std::vector<RangeIndexEntry> const& get_range_index() const noexcept { return m_range_index; }
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
std::vector<RangeIndexEntry> const& get_range_index() const { return m_range_index; }
[[nodiscard]] std::vector<RangeIndexEntry> const& get_range_index() const noexcept { return m_range_index; }
🤖 Prompt for AI Agents
In components/core/src/clp_s/ArchiveReaderAdaptor.hpp at line 90, the getter
method get_range_index() should be updated to include the [[nodiscard]]
attribute and be marked noexcept. This change ensures consistency with other
getter methods and signals to callers that the return value should not be
ignored and that the method does not throw exceptions. Modify the method
signature to add [[nodiscard]] before the return type and noexcept after the
method declaration.


private:
/**
* Tries to read an ArchiveFileInfo packet from the archive metadata.
Expand Down Expand Up @@ -90,6 +114,14 @@ class ArchiveReaderAdaptor {
*/
ErrorCode try_read_archive_info(ZstdDecompressor& decompressor, size_t size);

/**
* Tries to read a RangeIndex packet from the archive metadata.
* @param decompressor
* @param size The number of decompressed bytes making up the packet.
* @return ErrorCodeSuccess on success or the relevant ErrorCode on failure.
*/
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Resolved
auto try_read_range_index(ZstdDecompressor& decompressor, size_t size) -> ErrorCode;

/**
* Tries to read an unknown metadata packet from the archive metadata.
* @param decompressor
Expand Down Expand Up @@ -140,6 +172,7 @@ class ArchiveReaderAdaptor {
std::optional<std::string> m_current_reader_holder;
std::shared_ptr<TimestampDictionaryReader> m_timestamp_dictionary;
std::shared_ptr<clp::ReaderInterface> m_reader;
std::vector<RangeIndexEntry> m_range_index;
};
} // namespace clp_s
#endif // CLP_S_ARCHIVEREADERADAPTOR_HPP
1 change: 1 addition & 0 deletions components/core/src/clp_s/indexer/CMakeLists.txt
Original file line number Diff line number Diff line change
Expand Up @@ -90,6 +90,7 @@ target_link_libraries(indexer
clp::string_utils
date::date
MariaDBClient::MariaDBClient
nlohmann_json::nlohmann_json
OpenSSL::Crypto
simdjson::simdjson
spdlog::spdlog
Expand Down
Loading