Merge pull request #26090 from WillemKauf/manual-backport-26080-v24.1.x-556

[v24.1.x] `storage`: fix `non_data_timestamps` with overflow (Manual backport)

Author: piyushredpanda

src/v/storage/index_state.cc

@@ -59,7 +59,16 @@ bool index_state::maybe_index(
     // to override the timestamps of any config batch that was indexed
     // by virtue of being the first in the segment.
     if (user_data && non_data_timestamps) {
-        vassert(relative_time_index.size() == 1, "");
+        // We can only add a non-data timestamp to the empty index. This
+        // is why we can assume that the index size is 1.
+        // Otherwise it will be impossible to to reset the non-data timestamp.
+        vassert(
+          relative_time_index.size() == 1,
+          "Relative time index can not be reset, unexpected index size {} "
+          "(expected 1). This can only happen if more than one non-data "
+          "timestamp was added to the index: {}.",
+          relative_time_index.size(),
+          *this);
         relative_time_index[0]
           = offset_time_index{last_timestamp, with_offset}.raw_value();
 
@@ -70,14 +79,15 @@ bool index_state::maybe_index(
 
     // index_state
     bool is_empty = false;
+    bool should_set_non_data_timestamp = false;
     if (empty()) {
         // Ordinarily, we do not allow configuration batches to contribute to
         // the segment's timestamp bounds (because config batches use walltime
         // but user data timestamps may be anything).  However, for the first
         // batch we set the timestamps, and then set a `non_data_timestamps`
         // flag so that the next time we see user data we will overwrite
         // the walltime timestamps with the user data timestamps.
-        non_data_timestamps = !user_data;
+        should_set_non_data_timestamp = !user_data;
 
         base_timestamp = first_timestamp;
         max_timestamp = first_timestamp;
@@ -113,7 +123,9 @@ bool index_state::maybe_index(
               batch_base_offset() - base_offset(),
               offset_time_index{last_timestamp - base_timestamp, with_offset},
               starting_position_in_file);
-
+            if (should_set_non_data_timestamp) {
+                non_data_timestamps = true;
+            }
             return true;
         } else {
             // We can't index anything beyond uint32 space. Presumably no

src/v/storage/tests/index_state_test.cc

@@ -244,3 +244,34 @@ BOOST_AUTO_TEST_CASE(offset_time_index_test) {
         }
     }
 }
+
+BOOST_AUTO_TEST_CASE(non_data_timestamps_with_overflow) {
+    storage::index_state state;
+
+    // Need to ensure that non_data_timestamps in the segment_index is only set
+    // iff there is an entry in the index_state. Otherwise, a call to
+    // index.try_reset_relative_time_index() will trigger a vassert.
+    constexpr long uint32_max = std::numeric_limits<uint32_t>::max();
+    state.maybe_index(
+      0,
+      1,
+      0,
+      model::offset{uint32_max + 1},
+      model::offset{uint32_max + 1},
+      model::timestamp{0},
+      model::timestamp{0},
+      std::nullopt,
+      false,
+      0);
+    state.maybe_index(
+      1,
+      1,
+      1,
+      model::offset{uint32_max + 2},
+      model::offset{uint32_max + 2},
+      model::timestamp{0},
+      model::timestamp{0},
+      std::nullopt,
+      true,
+      0);
+}