diff --git a/src/v/lsm/block/BUILD b/src/v/lsm/block/BUILD index 4f8b582bc26b6..b221bfc2e8745 100644 --- a/src/v/lsm/block/BUILD +++ b/src/v/lsm/block/BUILD @@ -66,7 +66,10 @@ redpanda_cc_library( "filter.h", ], implementation_deps = [ + "//src/v/hashing:crc32c", "//src/v/hashing:xx", + "//src/v/lsm/core:compression", + "//src/v/lsm/core:exceptions", ], deps = [ ":contents", diff --git a/src/v/lsm/block/filter.cc b/src/v/lsm/block/filter.cc index b852cd4a603b0..2ba0831c1d9e2 100644 --- a/src/v/lsm/block/filter.cc +++ b/src/v/lsm/block/filter.cc @@ -10,10 +10,15 @@ #include "base/units.h" #include "base/vassert.h" +#include "hashing/crc32c.h" #include "hashing/xx.h" +#include "lsm/core/compression.h" +#include "lsm/core/exceptions.h" #include "lsm/core/internal/keys.h" #include +#include +#include #include #include @@ -150,44 +155,172 @@ void filter_builder::generate_filter() { _keys.clear(); } -filter_reader::filter_reader(ss::lw_shared_ptr c) - : _contents(std::move(c)) - , _offset(0) - , _num(0) - , _base_lg(0) { - size_t n = _contents->size(); - constexpr static size_t min_footer_size = sizeof(uint8_t) - + sizeof(uint32_t); - if (n < min_footer_size) { - return; +filter_reader::filter_reader( + io::random_access_file_reader* file, + uint64_t content_offset, + contents tail, + uint32_t offsets_start, + size_t num, + uint8_t base_lg, + chunked_vector region_crcs) + : _file(file) + , _content_offset(content_offset) + , _tail(std::move(tail)) + , _offsets_start(offsets_start) + , _num(num) + , _base_lg(base_lg) + , _region_crcs(std::move(region_crcs)) {} + +ss::future filter_reader::open( + io::random_access_file_reader* file, + uint64_t content_offset, + uint64_t content_size) { + // The content ends with a (offsets_start: u32, base_lg: u8) trailer. + constexpr size_t trailer_size = sizeof(uint32_t) + sizeof(uint8_t); + if (content_size < trailer_size) { + // Too small to be a valid filter; treat every key as "may match". + co_return filter_reader( + file, content_offset, contents(ioarray{}), 0, 0, 0, {}); } - _base_lg = (*_contents)[n - 1]; - auto offsets_start = _contents->read_fixed32(n - min_footer_size); - if (offsets_start > n - min_footer_size) { - return; + auto trailer = co_await file->read( + content_offset + content_size - trailer_size, trailer_size); + uint32_t offsets_start = trailer.read_fixed32(0); + auto base_lg = static_cast(trailer[sizeof(uint32_t)]); + if (offsets_start > content_size - trailer_size) { + co_return filter_reader( + file, + content_offset, + contents(ioarray{}), + offsets_start, + 0, + base_lg, + {}); + } + size_t num = (content_size - trailer_size - offsets_start) + / sizeof(uint32_t); + // Retain only the tail: the offset array plus the trailer. The bulk of the + // filter (the per-region bloom data) is left on disk and read on demand. + auto tail_buf = co_await file->read( + content_offset + offsets_start, content_size - offsets_start); + block::contents tail(std::move(tail_buf)); + + // The SST builder writes the filter as an uncompressed block, so a trailer + // -- a compression-type byte followed by the whole-block CRC -- immediately + // follows the content. Validate that CRC, and in the same pass compute a + // CRC over each region so lazy region reads can be checked against it. + constexpr size_t block_trailer_size = sizeof(uint8_t) + + sizeof(crc::crc32c::value_type); + auto block_trailer = co_await file->read( + content_offset + content_size, block_trailer_size); + if ( + compression_type_from_raw(static_cast(block_trailer[0])) + != compression_type::none) { + co_await ss::coroutine::return_exception(corruption_exception( + "filter block unexpectedly compressed in {}", *file)); + } + uint32_t expected_crc = block_trailer.read_fixed32(sizeof(uint8_t)); + + // Region ordinal i covers [offset[i], offset[i+1]); the last region ends at + // offsets_start. Offsets are contiguous, so a single forward scan over the + // region data assigns each byte to exactly one region. + auto region_end = [&](size_t i) -> uint64_t { + return (i + 1 < num) ? tail.read_fixed32((i + 1) * sizeof(uint32_t)) + : offsets_start; + }; + constexpr size_t crc_window = 128 * 1024; + crc::crc32c whole; + chunked_vector region_crcs; + region_crcs.reserve(num); + size_t region_idx = 0; + crc::crc32c region; + for (uint64_t rel = 0; rel < offsets_start;) { + uint64_t win_len = std::min(crc_window, offsets_start - rel); + auto buf = co_await file->read(content_offset + rel, win_len); + block::contents window(std::move(buf)); + crc_extend_iobuf(whole, window.share(0, win_len)); + // Split this window at region boundaries, carrying a region's CRC + // across windows when the region is larger than one. + for (uint64_t cur = rel; cur < rel + win_len && region_idx < num;) { + uint64_t end = std::min( + region_end(region_idx), rel + win_len); + if (end > cur) { + crc_extend_iobuf(region, window.share(cur - rel, end - cur)); + cur = end; + } + if (cur == region_end(region_idx)) { + region_crcs.push_back(region.value()); + region = crc::crc32c{}; + ++region_idx; + } else { + break; // region continues into the next window + } + } + rel += win_len; } - _offset = offsets_start; - _num = (n - min_footer_size - offsets_start) / 4; + // Finalize any trailing empty regions (offset[i] == offsets_start). + while (region_idx < num) { + region_crcs.push_back(region.value()); + region = crc::crc32c{}; + ++region_idx; + } + + // Fold the retained tail and the compression byte into the whole-block CRC + // (it covers the content plus that byte) and verify it. + crc_extend_iobuf(whole, tail.share(0, content_size - offsets_start)); + char compression_byte = block_trailer[0]; + whole.extend(&compression_byte, 1); + if (whole.value() != expected_crc) { + co_await ss::coroutine::return_exception(corruption_exception( + "unexpected filter crc, got: {}, want: {} in {}", + whole.value(), + expected_crc, + *file)); + } + + co_return filter_reader( + file, + content_offset, + std::move(tail), + offsets_start, + num, + base_lg, + std::move(region_crcs)); } -bool filter_reader::key_may_match( - uint64_t block_offset, internal::key_view key) { +ss::future +filter_reader::key_may_match(uint64_t block_offset, internal::key_view key) { uint64_t index = block_offset >> _base_lg; - if (index < _num) { - // TODO(perf): This is very much on the hot path, and decoding int32 - // from segmented bytes is not going to be the cheapest operation in the - // world, maybe it makes sense to be able to decode these at - // construction time and free the memory in _contents, with some kind of - // `trim` operation. - uint32_t position = _offset + (index * sizeof(uint32_t)); - auto start = _contents->read_fixed32(position); - auto limit = _contents->read_fixed32(position + sizeof(uint32_t)); - if (start <= limit && limit <= _offset) { - return bloom_filter::key_may_match( - key.user_key(), *_contents, start, limit - start); - } + if (index >= _num) { + co_return true; // Unknown region: conservatively may match. + } + // The offset array lives at the start of the retained tail. Entry `index` + // is region `index`'s start (a position within the filter content); the + // next entry -- or the offsets_start trailer for the last region -- is its + // end. + uint32_t start = _tail.read_fixed32(index * sizeof(uint32_t)); + uint32_t limit = _tail.read_fixed32((index + 1) * sizeof(uint32_t)); + if (start > limit || limit > _offsets_start) { + co_return true; // Malformed offsets: conservatively may match. + } + if (limit - start < 2) { + co_return false; // Empty region: no keys hashed here. + } + auto region = co_await _file->read(_content_offset + start, limit - start); + block::contents region_contents(std::move(region)); + // Verify the region against the CRC computed at open, so a post-open disk + // bit-flip is caught here rather than silently becoming a false negative. + crc::crc32c actual; + crc_extend_iobuf(actual, region_contents.share(0, limit - start)); + if (actual.value() != _region_crcs[index]) { + co_await ss::coroutine::return_exception(corruption_exception( + "filter region {} failed crc check, got: {}, want: {} in {}", + index, + actual.value(), + _region_crcs[index], + *_file)); } - return true; // Unknown/error cases we don't know + co_return bloom_filter::key_may_match( + key.user_key(), region_contents, 0, limit - start); } } // namespace lsm::block diff --git a/src/v/lsm/block/filter.h b/src/v/lsm/block/filter.h index 0c3b268fae54b..71361eb0f72af 100644 --- a/src/v/lsm/block/filter.h +++ b/src/v/lsm/block/filter.h @@ -14,6 +14,7 @@ #include "lsm/block/contents.h" #include "lsm/core/internal/keys.h" +#include #include #include @@ -52,19 +53,61 @@ class filter_builder { }; // A reader for a filter block in an SST. +// +// The filter is read lazily to bound resident memory: a deep-level SST's filter +// is O(GiB), so holding it all (one per open reader) doesn't scale. open() +// retains only the per-region offset array and a per-region CRC array (together +// ~0.4% of the file); each key_may_match reads just the covering ~filter_period +// region from the file on demand. +// +// open() streams the whole block once to validate its CRC, and in that same +// pass computes a CRC over each region. The per-region CRCs are retained and +// checked on every lazy read, so a bit that rots on the (cache) disk after open +// is caught rather than silently turned into a bloom false negative -- the +// integrity of a lazily-read region is verified against a checksum established +// over bytes the whole-block CRC just proved intact. class filter_reader { public: - explicit filter_reader(ss::lw_shared_ptr); + /// Open a filter over the filter block content at + /// [content_offset, content_offset + content_size) in `file`, immediately + /// followed by the block trailer (a compression-type byte and the + /// whole-block CRC that open() validates). Retains the region offset array + /// and a per-region CRC array. `file` must outlive the returned reader. + static ss::future open( + io::random_access_file_reader* file, + uint64_t content_offset, + uint64_t content_size); // Check if it's possible that the user's key exists in the block at this - // offset within the SST. - bool key_may_match(uint64_t block_offset, internal::key_view key); + // offset within the SST. Reads the covering filter region from the file and + // verifies it against the region CRC retained at open; throws + // corruption_exception on a mismatch. + ss::future + key_may_match(uint64_t block_offset, internal::key_view key); private: - ss::lw_shared_ptr _contents; - size_t _offset; // The offset at which the data ends - size_t _num; + filter_reader( + io::random_access_file_reader* file, + uint64_t content_offset, + contents tail, + uint32_t offsets_start, + size_t num, + uint8_t base_lg, + chunked_vector region_crcs); + + io::random_access_file_reader* _file; + // File offset of the start of the filter block content. + uint64_t _content_offset; + // The retained tail of the filter block: the per-region offset array + // followed by the (offsets_start, base_lg) trailer. Region byte ranges + // (positions within the content) are read from here. + contents _tail; + uint32_t _offsets_start; // Content position where the offset array begins. + size_t _num; // Number of filter regions. uint8_t _base_lg; + // CRC32C of each region's bytes, computed at open over the validated block. + // Indexed by the same region ordinal as the offset array. + chunked_vector _region_crcs; }; } // namespace lsm::block diff --git a/src/v/lsm/block/tests/BUILD b/src/v/lsm/block/tests/BUILD index f0d9d13cb074f..3f6ba535f6ac3 100644 --- a/src/v/lsm/block/tests/BUILD +++ b/src/v/lsm/block/tests/BUILD @@ -26,10 +26,13 @@ redpanda_cc_gtest( ], deps = [ "//src/v/base", + "//src/v/bytes:ioarray", "//src/v/bytes:iobuf", - "//src/v/lsm/block:contents", + "//src/v/hashing:crc32c", "//src/v/lsm/block:filter", + "//src/v/lsm/core:exceptions", "//src/v/lsm/core/internal:keys", + "//src/v/lsm/io:persistence", "//src/v/test_utils:gtest", "@googletest//:gtest", "@seastar", diff --git a/src/v/lsm/block/tests/filter_test.cc b/src/v/lsm/block/tests/filter_test.cc index 45d9cc905866a..0cf834cba80ae 100644 --- a/src/v/lsm/block/tests/filter_test.cc +++ b/src/v/lsm/block/tests/filter_test.cc @@ -4,21 +4,87 @@ // https://github.com/google/leveldb/blob/main/AUTHORS for names of // contributors. // -// Modifications copyright 2025 Redpanda Data, Inc. +// Modifications copyright 2026 Redpanda Data, Inc. #include "base/seastarx.h" -#include "lsm/block/contents.h" +#include "bytes/ioarray.h" +#include "bytes/iobuf.h" +#include "hashing/crc32c.h" #include "lsm/block/filter.h" +#include "lsm/core/exceptions.h" #include "lsm/core/internal/keys.h" +#include "lsm/io/persistence.h" +#include "test_utils/test.h" -#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include namespace { +// Serves a fixed in-memory buffer as a random-access file, so the lazy +// filter_reader can read its regions without a real cache/cloud backend. +class in_memory_reader final : public lsm::io::random_access_file_reader { +public: + explicit in_memory_reader(iobuf data) + : _data(std::move(data)) {} + + ss::future read(size_t offset, size_t n) override { + n = std::min(n, _data.size_bytes() - offset); + co_return ioarray::copy_from(_data.share(offset, n)); + } + ss::future<> close() override { return ss::now(); } + fmt::iterator format_to(fmt::iterator it) const override { + return fmt::format_to(it, "in_memory_reader"); + } + + // Flip every bit of the byte at `offset`, simulating on-disk corruption + // that occurs after the filter has been opened. + void corrupt_byte(size_t offset) { + std::string buf(_data.size_bytes(), '\0'); + auto* p = buf.data(); + iobuf::iterator_consumer(_data.cbegin(), _data.cend()) + .consume(buf.size(), [&p](const char* src, size_t sz) { + std::memcpy(p, src, sz); + p += sz; + return ss::stop_iteration::no; + }); + buf[offset] = static_cast(~static_cast(buf[offset])); + iobuf out; + out.append(buf.data(), buf.size()); + _data = std::move(out); + } + +private: + iobuf _data; +}; + using key_vector = std::vector; using keys_by_block = std::map; -lsm::block::filter_reader make_filter(const keys_by_block& keys) { +// A filter_reader plus the file backing it, kept together so the file outlives +// the reader (which holds a non-owning pointer to it). +struct filter_under_test { + std::unique_ptr file; + lsm::block::filter_reader reader; + + ss::future + key_may_match(uint64_t block_offset, lsm::internal::key_view key) { + return reader.key_may_match(block_offset, key); + } +}; + +ss::future make_filter(const keys_by_block& keys) { lsm::block::filter_builder builder({}); for (const auto& [block, keys_in_block] : keys) { builder.start_block(block); @@ -27,76 +93,110 @@ lsm::block::filter_reader make_filter(const keys_by_block& keys) { lsm::internal::key::encode({.key = lsm::user_key_view(key)})); } } - auto c = lsm::block::contents::copy_from(builder.finish()); - return lsm::block::filter_reader(std::move(c)); + iobuf content = builder.finish(); + auto size = content.size_bytes(); + // Append the block trailer the SST builder writes: an (uncompressed) + // compression-type byte followed by a CRC over the content plus that byte. + // filter_reader::open validates this and derives the per-region CRCs. + content.append(std::to_array({0})); // compression_type::none + crc::crc32c crc; + crc_extend_iobuf(crc, content); + content.append( + std::bit_cast>(crc.value())); + auto file = std::make_unique(std::move(content)); + auto reader = co_await lsm::block::filter_reader::open( + file.get(), /*content_offset=*/0, size); + co_return filter_under_test{std::move(file), std::move(reader)}; } using lsm::internal::operator""_key; } // namespace -TEST(Filter, Empty) { - auto filter = make_filter({}); - ASSERT_TRUE(filter.key_may_match(0, "foo"_key)); - ASSERT_TRUE(filter.key_may_match(100000, "foo"_key)); +class FilterTest : public seastar_test {}; + +TEST_F(FilterTest, Empty) { + auto filter = make_filter({}).get(); + EXPECT_TRUE(filter.key_may_match(0, "foo"_key).get()); + EXPECT_TRUE(filter.key_may_match(100000, "foo"_key).get()); } -TEST(Filter, Bloom) { +TEST_F(FilterTest, Bloom) { auto reader = make_filter({ - {0, {"hello", "world"}}, - }); - EXPECT_TRUE(reader.key_may_match(0, "hello"_key)); - EXPECT_TRUE(reader.key_may_match(0, "world"_key)); - EXPECT_FALSE(reader.key_may_match(0, "x"_key)); - EXPECT_FALSE(reader.key_may_match(0, "foo"_key)); + {0, {"hello", "world"}}, + }) + .get(); + EXPECT_TRUE(reader.key_may_match(0, "hello"_key).get()); + EXPECT_TRUE(reader.key_may_match(0, "world"_key).get()); + EXPECT_FALSE(reader.key_may_match(0, "x"_key).get()); + EXPECT_FALSE(reader.key_may_match(0, "foo"_key).get()); } -TEST(Filter, SingleBlock) { +TEST_F(FilterTest, SingleBlock) { auto reader = make_filter({ - {100, {"foo", "bar", "box"}}, - {200, {"box"}}, - {300, {"hello"}}, - }); - EXPECT_TRUE(reader.key_may_match(100, "foo"_key)); - EXPECT_TRUE(reader.key_may_match(100, "bar"_key)); - EXPECT_TRUE(reader.key_may_match(100, "box"_key)); - EXPECT_TRUE(reader.key_may_match(100, "hello"_key)); - EXPECT_TRUE(reader.key_may_match(100, "foo"_key)); - EXPECT_FALSE(reader.key_may_match(100, "missing"_key)); - EXPECT_FALSE(reader.key_may_match(100, "other"_key)); + {100, {"foo", "bar", "box"}}, + {200, {"box"}}, + {300, {"hello"}}, + }) + .get(); + EXPECT_TRUE(reader.key_may_match(100, "foo"_key).get()); + EXPECT_TRUE(reader.key_may_match(100, "bar"_key).get()); + EXPECT_TRUE(reader.key_may_match(100, "box"_key).get()); + EXPECT_TRUE(reader.key_may_match(100, "hello"_key).get()); + EXPECT_TRUE(reader.key_may_match(100, "foo"_key).get()); + EXPECT_FALSE(reader.key_may_match(100, "missing"_key).get()); + EXPECT_FALSE(reader.key_may_match(100, "other"_key).get()); } -TEST(FilterBlockTest, MultipleBlocks) { +TEST_F(FilterTest, MultipleBlocks) { auto reader = make_filter({ - {0, {"foo"}}, - {2000, {"bar"}}, - {3100, {"box"}}, - {9000, {"box", "hello"}}, - }); + {0, {"foo"}}, + {2000, {"bar"}}, + {3100, {"box"}}, + {9000, {"box", "hello"}}, + }) + .get(); // Check first filter - EXPECT_TRUE(reader.key_may_match(0, "foo"_key)); - EXPECT_TRUE(reader.key_may_match(2000, "bar"_key)); - EXPECT_FALSE(reader.key_may_match(0, "box"_key)); - EXPECT_FALSE(reader.key_may_match(0, "hello"_key)); - EXPECT_FALSE(reader.key_may_match(0, "world"_key)); + EXPECT_TRUE(reader.key_may_match(0, "foo"_key).get()); + EXPECT_TRUE(reader.key_may_match(2000, "bar"_key).get()); + EXPECT_FALSE(reader.key_may_match(0, "box"_key).get()); + EXPECT_FALSE(reader.key_may_match(0, "hello"_key).get()); + EXPECT_FALSE(reader.key_may_match(0, "world"_key).get()); // Check second filter - EXPECT_TRUE(reader.key_may_match(3100, "box"_key)); - EXPECT_FALSE(reader.key_may_match(3100, "foo"_key)); - EXPECT_FALSE(reader.key_may_match(3100, "bar"_key)); - EXPECT_FALSE(reader.key_may_match(3100, "hello"_key)); - EXPECT_FALSE(reader.key_may_match(3100, "world"_key)); + EXPECT_TRUE(reader.key_may_match(3100, "box"_key).get()); + EXPECT_FALSE(reader.key_may_match(3100, "foo"_key).get()); + EXPECT_FALSE(reader.key_may_match(3100, "bar"_key).get()); + EXPECT_FALSE(reader.key_may_match(3100, "hello"_key).get()); + EXPECT_FALSE(reader.key_may_match(3100, "world"_key).get()); // Check third filter (empty) - EXPECT_FALSE(reader.key_may_match(4100, "foo"_key)); - EXPECT_FALSE(reader.key_may_match(4100, "bar"_key)); - EXPECT_FALSE(reader.key_may_match(4100, "box"_key)); - EXPECT_FALSE(reader.key_may_match(4100, "hello"_key)); + EXPECT_FALSE(reader.key_may_match(4100, "foo"_key).get()); + EXPECT_FALSE(reader.key_may_match(4100, "bar"_key).get()); + EXPECT_FALSE(reader.key_may_match(4100, "box"_key).get()); + EXPECT_FALSE(reader.key_may_match(4100, "hello"_key).get()); // Check last filter - EXPECT_TRUE(reader.key_may_match(9000, "box"_key)); - EXPECT_TRUE(reader.key_may_match(9000, "hello"_key)); - EXPECT_FALSE(reader.key_may_match(9000, "foo"_key)); - EXPECT_FALSE(reader.key_may_match(9000, "bar"_key)); + EXPECT_TRUE(reader.key_may_match(9000, "box"_key).get()); + EXPECT_TRUE(reader.key_may_match(9000, "hello"_key).get()); + EXPECT_FALSE(reader.key_may_match(9000, "foo"_key).get()); + EXPECT_FALSE(reader.key_may_match(9000, "bar"_key).get()); +} + +// A bit that rots in a region's bloom data after open must be caught on the +// lazy read via the per-region CRC, not silently returned as a false negative. +TEST_F(FilterTest, DetectsRegionCorruptionAfterOpen) { + auto filter = make_filter({ + {0, {"hello", "world"}}, + }) + .get(); + // A clean read succeeds and the open-time whole-block CRC validated. + EXPECT_TRUE(filter.key_may_match(0, "hello"_key).get()); + + // Corrupt the first byte of region 0's bloom data on "disk" after open. + filter.file->corrupt_byte(0); + + EXPECT_THROW( + filter.key_may_match(0, "hello"_key).get(), lsm::corruption_exception); } diff --git a/src/v/lsm/sst/reader.cc b/src/v/lsm/sst/reader.cc index 13ab60c22bdff..4ed7626b823a2 100644 --- a/src/v/lsm/sst/reader.cc +++ b/src/v/lsm/sst/reader.cc @@ -75,8 +75,10 @@ ss::future> read_filter( co_return std::nullopt; } auto filter_handle = block::handle::from_iobuf(iter->value()); - auto filter_contents = co_await read_block(file, filter_handle); - co_return block::filter_reader(std::move(filter_contents)); + // filter_reader::open validates the block's CRC (and retains per-region + // CRCs) while reading only the tail, so no separate whole-block pass here. + co_return co_await block::filter_reader::open( + file, filter_handle.offset, filter_handle.size); } } // namespace @@ -125,7 +127,7 @@ class reader::impl { auto v = iiter->value(); if (_filter) { auto handle = block::handle::from_iobuf(v.share()); - if (!_filter->key_may_match(handle.offset, key)) { + if (!co_await _filter->key_may_match(handle.offset, key)) { // Bloom filter says it's certainly not there. co_return; } diff --git a/src/v/lsm/sst/tests/BUILD b/src/v/lsm/sst/tests/BUILD index 4dd760d8932df..516f2f1f23a0d 100644 --- a/src/v/lsm/sst/tests/BUILD +++ b/src/v/lsm/sst/tests/BUILD @@ -29,12 +29,20 @@ redpanda_cc_gtest( ], deps = [ "//src/v/base", + "//src/v/bytes:ioarray", "//src/v/bytes:iobuf", + "//src/v/bytes:iobuf_parser", + "//src/v/lsm/block:contents", + "//src/v/lsm/block:handle", + "//src/v/lsm/block:reader", + "//src/v/lsm/core:exceptions", "//src/v/lsm/core/internal:keys", "//src/v/lsm/io:disk_persistence", "//src/v/lsm/io:memory_persistence", + "//src/v/lsm/io:persistence", "//src/v/lsm/sst:block_cache", "//src/v/lsm/sst:builder", + "//src/v/lsm/sst:footer", "//src/v/lsm/sst:reader", "//src/v/test_utils:gtest", "//src/v/utils:uuid", diff --git a/src/v/lsm/sst/tests/sst_test.cc b/src/v/lsm/sst/tests/sst_test.cc index caf89c51a1ba5..16fb86e15d111 100644 --- a/src/v/lsm/sst/tests/sst_test.cc +++ b/src/v/lsm/sst/tests/sst_test.cc @@ -7,19 +7,58 @@ // Modifications copyright 2025 Redpanda Data, Inc. #include "base/seastarx.h" +#include "bytes/ioarray.h" +#include "bytes/iobuf.h" +#include "bytes/iobuf_parser.h" +#include "lsm/block/contents.h" +#include "lsm/block/handle.h" +#include "lsm/block/reader.h" +#include "lsm/core/exceptions.h" #include "lsm/core/internal/keys.h" #include "lsm/io/disk_persistence.h" #include "lsm/io/memory_persistence.h" +#include "lsm/io/persistence.h" #include "lsm/sst/builder.h" +#include "lsm/sst/footer.h" #include "lsm/sst/reader.h" #include "utils/uuid.h" +#include +#include + #include +#include +#include + using lsm::internal::operator""_key; using persistence_factory = std::function>()>; +namespace { + +// Serves a fixed in-memory buffer as a random-access file, so a deliberately +// corrupted copy of an SST can be opened. +class in_memory_reader final : public lsm::io::random_access_file_reader { +public: + explicit in_memory_reader(iobuf data) + : _data(std::move(data)) {} + + ss::future read(size_t offset, size_t n) override { + n = std::min(n, _data.size_bytes() - offset); + co_return ioarray::copy_from(_data.share(offset, n)); + } + ss::future<> close() override { return ss::now(); } + fmt::iterator format_to(fmt::iterator it) const override { + return fmt::format_to(it, "in_memory_reader"); + } + +private: + iobuf _data; +}; + +} // namespace + class SSTTest : public ::testing::TestWithParam { void SetUp() override { _persistence = GetParam()().get(); } void TearDown() override { @@ -67,6 +106,67 @@ TEST_P(SSTTest, CanCreateAndReadFile) { reader.close().get(); } +// Corrupting a byte in the filter block must be caught by the streaming CRC +// validation in reader::open, since the lazy filter reader no longer holds (and +// re-checks) the whole filter in memory. Proves integrity is preserved. +TEST(SSTCorruption, FilterCrcMismatchCaughtOnOpen) { + auto persistence = lsm::io::make_memory_data_persistence(); + auto persistence_cleanup = ss::defer([&] { persistence->close().get(); }); + + size_t file_size = 0; + { + auto file = persistence->open_sequential_writer({}).get(); + lsm::sst::builder builder(std::move(file), {}); + builder.add("abc"_key, iobuf::from("value1")).get(); + builder.add("abcd"_key, iobuf::from("value2")).get(); + builder.add("abc123"_key, iobuf::from("value3")).get(); + builder.finish().get(); + builder.close().get(); + file_size = builder.file_size(); + } + + auto rdr_opt = persistence->open_random_access_reader({}, file_size).get(); + ASSERT_TRUE(bool(rdr_opt)); + auto& rdr = *rdr_opt; + auto rdr_cleanup = ss::defer([&] { rdr->close().get(); }); + + // Locate the filter block: footer -> metaindex -> "filter.*" handle. + auto footer_bytes = rdr + ->read( + file_size - lsm::sst::footer::encoded_length, + lsm::sst::footer::encoded_length) + .get(); + auto footer = lsm::sst::footer::from_iobuf(footer_bytes.as_iobuf()); + auto meta + = lsm::block::contents::read(rdr.get(), footer.metaindex_handle).get(); + lsm::block::reader metaindex(std::move(meta)); + auto it = metaindex.create_iterator(); + auto filter_key = lsm::internal::key::encode( + {.key = lsm::user_key_view("filter.RedpandaBloomV0")}); + it->seek(filter_key).get(); + ASSERT_TRUE(it->valid()); + auto filter_handle = lsm::block::handle::from_iobuf(it->value()); + + // Read the whole file, flip one byte inside the filter block, and reopen. + auto whole = rdr->read(0, file_size).get(); + iobuf_parser p(whole.as_iobuf()); + auto bytes = p.read_string_unsafe(p.bytes_left()); + ASSERT_LT(filter_handle.offset, bytes.size()); + bytes[filter_handle.offset] = static_cast( + bytes[filter_handle.offset] ^ 0xFF); + + auto corrupt = std::make_unique(iobuf::from(bytes)); + EXPECT_THROW( + lsm::sst::reader::open( + std::move(corrupt), + lsm::internal::file_id{0}, + file_size, + ss::make_lw_shared( + 1_MiB, ss::make_lw_shared())) + .get(), + lsm::corruption_exception); +} + INSTANTIATE_TEST_SUITE_P( SSTTestSuite, SSTTest,