Merge pull request redpanda-data#26051 from andrwng/storage-index-overflow-truncate

storage: fix index state truncate overflow

Author: andrwng

src/v/storage/index_state.cc

@@ -772,12 +772,21 @@ bool index_state::truncate(
     if (new_max_offset < base_offset) {
         return needs_persistence;
     }
-    const uint32_t i = new_max_offset() - base_offset();
-    auto res = index.offset_lower_bound(i);
-    size_t remove_back_elems = index.size() - res.value_or(index.size());
-    if (remove_back_elems > 0) {
-        needs_persistence = true;
-        pop_back(remove_back_elems);
+    static constexpr int64_t u32_max = std::numeric_limits<uint32_t>::max();
+    int64_t delta = new_max_offset() - base_offset();
+    // NOTE: we expect that deltas above u32_max would have not been
+    // added to the index, unless by an older buggy version of Redpanda.
+    // With this in mind, either there are no entries above u32_max, or
+    // offset_lower_bound() isn't going to work correctly anyway, so we just
+    // skip removal and rely on queries to detect the overflow.
+    if (delta <= u32_max) {
+        auto i = static_cast<uint32_t>(delta);
+        auto res = index.offset_lower_bound(i);
+        size_t remove_back_elems = index.size() - res.value_or(index.size());
+        if (remove_back_elems > 0) {
+            needs_persistence = true;
+            pop_back(remove_back_elems);
+        }
     }
     if (new_max_offset < max_offset) {
         needs_persistence = true;

src/v/storage/tests/index_state_test.cc

@@ -410,3 +410,36 @@ TEST(IndexState, NonDataTimestampsWithOverflow) {
       true,
       0);
 }
+
+BOOST_AUTO_TEST_CASE(index_overflow_truncate) {
+    storage::index_state state;
+
+    // Previous versions of Redpanda can have an index with offsets spanning
+    // greater than uint32 offset space. In these cases, the index is not
+    // reliable and truncation shouldn't attempt to truncate entries based on
+    // an overflown lookup. But, queries should fallback to returning the
+    // beginning of the index.
+    const model::timestamp dummy_ts;
+    const storage::offset_delta_time should_offset{false};
+    storage::offset_time_index time_idx{dummy_ts, should_offset};
+    constexpr long uint32_max = std::numeric_limits<uint32_t>::max();
+    state.add_entry(0, time_idx, 1);
+    state.add_entry(100, time_idx, 2);
+    state.add_entry(static_cast<uint32_t>(uint32_max + 1), time_idx, 3);
+    state.add_entry(static_cast<uint32_t>(uint32_max + 10), time_idx, 4);
+    state.max_offset = model::offset{uint32_max + 10};
+    BOOST_CHECK_EQUAL(4, state.size());
+
+    // The truncation shouldn't remove index entries, given the overflow.
+    auto needs_flush = state.truncate(
+      model::offset(uint32_max + 1), model::timestamp::now());
+    BOOST_CHECK(needs_flush);
+    BOOST_CHECK_EQUAL(4, state.size());
+    BOOST_CHECK_EQUAL(state.max_offset, model::offset{uint32_max + 1});
+
+    // Queries for the offset should start from the beginning of the segment.
+    auto res = state.find_nearest(model::offset(uint32_max + 1));
+    BOOST_REQUIRE(res.has_value());
+    BOOST_CHECK_EQUAL(res->offset, model::offset{0});
+    BOOST_CHECK_EQUAL(res->filepos, 1);
+}