From b588c76634b780e36084de1925063bf22bb9391c Mon Sep 17 00:00:00 2001 From: Shitao Weng Date: Tue, 18 Aug 2026 09:31:19 -0700 Subject: [PATCH] fix(stream): bound range reads by metadata entries --- src/types/redis_stream.cc | 61 +++++++++++++++++++++------- tests/cppunit/types/stream_test.cc | 65 ++++++++++++++++++++++++++++++ 2 files changed, 112 insertions(+), 14 deletions(-) diff --git a/src/types/redis_stream.cc b/src/types/redis_stream.cc index 66e05bbda06..0cb12740da0 100644 --- a/src/types/redis_stream.cc +++ b/src/types/redis_stream.cc @@ -1151,11 +1151,47 @@ rocksdb::Status Stream::Len(engine::Context &ctx, const Slice &stream_name, cons rocksdb::Status Stream::range(engine::Context &ctx, const std::string &ns_key, const StreamMetadata &metadata, const StreamRangeOptions &options, std::vector *entries) const { - std::string start_key = internalKeyFromEntryID(ns_key, metadata, options.start); - std::string end_key = internalKeyFromEntryID(ns_key, metadata, options.end); + if (metadata.size == 0) { + return rocksdb::Status::OK(); + } + + if ((!options.reverse && options.end < options.start) || (options.reverse && options.start < options.end)) { + return rocksdb::Status::OK(); + } + + // Metadata is the logical stream frontier; raw keys outside it may remain after an interrupted migration. + StreamRangeOptions range_options = options; + if (options.reverse) { + if (options.start < metadata.first_entry_id || options.end > metadata.last_entry_id) { + return rocksdb::Status::OK(); + } + if (options.start > metadata.last_entry_id) { + range_options.start = metadata.last_entry_id; + range_options.exclude_start = false; + } + if (options.end < metadata.first_entry_id) { + range_options.end = metadata.first_entry_id; + range_options.exclude_end = false; + } + } else { + if (options.end < metadata.first_entry_id || options.start > metadata.last_entry_id) { + return rocksdb::Status::OK(); + } + if (options.start < metadata.first_entry_id) { + range_options.start = metadata.first_entry_id; + range_options.exclude_start = false; + } + if (options.end > metadata.last_entry_id) { + range_options.end = metadata.last_entry_id; + range_options.exclude_end = false; + } + } + + std::string start_key = internalKeyFromEntryID(ns_key, metadata, range_options.start); + std::string end_key = internalKeyFromEntryID(ns_key, metadata, range_options.end); if (start_key == end_key) { - if (options.exclude_start || options.exclude_end) { + if (range_options.exclude_start || range_options.exclude_end) { return rocksdb::Status::OK(); } @@ -1171,11 +1207,7 @@ rocksdb::Status Stream::range(engine::Context &ctx, const std::string &ns_key, c return rocksdb::Status::InvalidArgument(rv.Msg()); } - entries->emplace_back(options.start.ToString(), std::move(values)); - return rocksdb::Status::OK(); - } - - if ((!options.reverse && options.end < options.start) || (options.reverse && options.start < options.end)) { + entries->emplace_back(range_options.start.ToString(), std::move(values)); return rocksdb::Status::OK(); } @@ -1191,20 +1223,21 @@ rocksdb::Status Stream::range(engine::Context &ctx, const std::string &ns_key, c auto iter = util::UniqueIterator(ctx, read_options, stream_cf_handle_); iter->Seek(start_key); - if (options.reverse && (!iter->Valid() || iter->key().ToString() != start_key)) { + if (range_options.reverse && (!iter->Valid() || iter->key().ToString() != start_key)) { iter->SeekForPrev(start_key); } - for (; iter->Valid() && (options.reverse ? iter->key().ToString() >= end_key : iter->key().ToString() <= end_key); - options.reverse ? iter->Prev() : iter->Next()) { + for (; + iter->Valid() && (range_options.reverse ? iter->key().ToString() >= end_key : iter->key().ToString() <= end_key); + range_options.reverse ? iter->Prev() : iter->Next()) { if (identifySubkeyType(iter->key()) != StreamSubkeyType::StreamEntry) { continue; } - if (options.exclude_start && iter->key().ToString() == start_key) { + if (range_options.exclude_start && iter->key().ToString() == start_key) { continue; } - if (options.exclude_end && iter->key().ToString() == end_key) { + if (range_options.exclude_end && iter->key().ToString() == end_key) { break; } @@ -1216,7 +1249,7 @@ rocksdb::Status Stream::range(engine::Context &ctx, const std::string &ns_key, c entries->emplace_back(entryIDFromInternalKey(iter->key()).ToString(), std::move(values)); - if (options.with_count && entries->size() == options.count) { + if (range_options.with_count && entries->size() == range_options.count) { break; } } diff --git a/tests/cppunit/types/stream_test.cc b/tests/cppunit/types/stream_test.cc index 9307095ccc5..bf07461cbd1 100644 --- a/tests/cppunit/types/stream_test.cc +++ b/tests/cppunit/types/stream_test.cc @@ -47,6 +47,22 @@ class RedisStreamTest : public TestBase { // NOLINT void SetUp() override { auto s = stream_->Del(*ctx_, name_); } void TearDown() override { auto s = stream_->Del(*ctx_, name_); } + rocksdb::Status PutRawEntry(const redis::StreamEntryID &id, const std::vector &values) { + std::string ns_key = stream_->AppendNamespacePrefix(name_); + StreamMetadata metadata(false); + auto s = stream_->GetMetadata(*ctx_, ns_key, &metadata); + if (!s.ok()) return s; + + std::string sub_key; + PutFixed64(&sub_key, id.ms); + PutFixed64(&sub_key, id.seq); + std::string entry_key = InternalKey(ns_key, sub_key, metadata.version, storage_->IsSlotIdEncoded()).Encode(); + auto batch = storage_->GetWriteBatchBase(); + s = batch->Put(storage_->GetCFHandle(ColumnFamilyID::Stream), entry_key, redis::EncodeStreamEntryValue(values)); + if (!s.ok()) return s; + return storage_->Write(*ctx_, storage_->DefaultWriteOptions(), batch->GetWriteBatch()); + } + std::string name_; redis::Stream *stream_; }; @@ -649,6 +665,55 @@ TEST_F(RedisStreamTest, RangeFromMinimumToMaximum) { CheckStreamEntryValues(entries[3].values, values4); } +TEST_F(RedisStreamTest, RangeIgnoresEntriesOutsideMetadataBounds) { + redis::StreamAddOptions add_options; + redis::StreamEntryID id; + for (const char *entry_id : {"1-0", "2-0", "3-0"}) { + add_options.next_id_strategy = *ParseNextStreamEntryIDStrategy(entry_id); + auto s = stream_->Add(*ctx_, name_, add_options, {"key", entry_id}, &id); + ASSERT_TRUE(s.ok()); + } + + redis::StreamTrimOptions trim_options; + trim_options.strategy = redis::StreamTrimStrategy::MinID; + trim_options.min_id = redis::StreamEntryID{2, 0}; + uint64_t trimmed = 0; + auto s = stream_->Trim(*ctx_, name_, trim_options, &trimmed); + ASSERT_TRUE(s.ok()); + ASSERT_EQ(trimmed, 1); + + ASSERT_TRUE(PutRawEntry({1, 0}, {"stale", "before"}).ok()); + ASSERT_TRUE(PutRawEntry({4, 0}, {"stale", "after"}).ok()); + + redis::StreamRangeOptions range_options; + range_options.start = redis::StreamEntryID::Minimum(); + range_options.end = redis::StreamEntryID::Maximum(); + std::vector entries; + s = stream_->Range(*ctx_, name_, range_options, &entries); + ASSERT_TRUE(s.ok()); + ASSERT_EQ(entries.size(), 2); + EXPECT_EQ(entries[0].key, "2-0"); + EXPECT_EQ(entries[1].key, "3-0"); + + range_options.reverse = true; + range_options.start = redis::StreamEntryID::Maximum(); + range_options.end = redis::StreamEntryID::Minimum(); + entries.clear(); + s = stream_->Range(*ctx_, name_, range_options, &entries); + ASSERT_TRUE(s.ok()); + ASSERT_EQ(entries.size(), 2); + EXPECT_EQ(entries[0].key, "3-0"); + EXPECT_EQ(entries[1].key, "2-0"); + + range_options.reverse = false; + range_options.start = {1, 0}; + range_options.end = {1, 0}; + entries.clear(); + s = stream_->Range(*ctx_, name_, range_options, &entries); + ASSERT_TRUE(s.ok()); + EXPECT_TRUE(entries.empty()); +} + TEST_F(RedisStreamTest, RangeFromMinimumToMinimum) { redis::StreamAddOptions add_options; add_options.next_id_strategy = *ParseNextStreamEntryIDStrategy(fmt::format("{}-{}", 123456, 1));