storage: use minimum of broker_timestamp and max_ts for retention_ms

The motivating case for `broker_time_based_retention` was the
fact that records with bad timestamps produced in the future
could lead to time-based retention being stuck indefinitely [1].

However, using _only_ the `broker_ts` can lead to unexpected
behavior when e.g. replicating data from an existing cluster
using MM2, as the timestamps of the Kafka records themselves are
correctly preserved, but internally, `redpanda` data structures
are not.

To avoid the potentially curious behavior of a divergence in
retention enforcement, take the minimum of the record's timestamp
as written and the current broker time. This will achieve the
original goal of preventing future timestamps from blocking
retention enforcement, while also avoiding any unexpected
behavior with past record timestamps.

[1]:
* redpanda-data#9820
* redpanda-data#12991

Author: WillemKauf

src/v/storage/segment_index.h

@@ -83,7 +83,23 @@ struct time_based_retention_cfg {
       std::optional<model::timestamp> alternative_retention_ts) const noexcept {
         // new clusters and new segments should hit this branch
         if (likely(use_broker_time && broker_ts.has_value())) {
-            return *broker_ts;
+            // The motivating case for `broker_time_based_retention` was the
+            // fact that records with bad timestamps produced in the future
+            // could lead to time-based retention being stuck indefinitely (See:
+            // https://github.com/redpanda-data/redpanda/issues/9820,
+            // https://github.com/redpanda-data/redpanda/pull/12991).
+            // However, using _only_ the `broker_ts` can lead to unexpected
+            // behavior when e.g. replicating data from an existing cluster
+            // using MM2, as the timestamps of the Kafka records themselves are
+            // correctly preserved, but internally, `redpanda` data structures
+            // are not.
+            // To avoid the potentially curious behavior of a divergence in
+            // retention enforcement, take the minimum of the record's timestamp
+            // as written and the current broker time. This will achieve the
+            // original goal of preventing future timestamps from blocking
+            // retention enforcement, while also avoiding any unexpected
+            // behavior with past record timestamps.
+            return std::min(max_ts, broker_ts.value());
         }
         // don't use broker time or no broker time available. fallback
         if (unlikely(

src/v/storage/tests/storage_e2e_test.cc

@@ -695,29 +695,8 @@ TEST_F(storage_test_fixture, test_time_based_eviction) {
             as);
       };
 
-    // gc with timestamp -1s, no segments should be evicted
+    // gc with timestamp -1s, all segments should be evicted
     compact_and_prefix_truncate(*disk_log, make_compaction_cfg(broker_t0 - 2s));
-    ASSERT_EQ(disk_log->segments().size(), 3);
-    ASSERT_EQ(
-      disk_log->segments().front()->offsets().get_base_offset(),
-      model::offset(0));
-    ASSERT_EQ(
-      disk_log->segments().back()->offsets().get_dirty_offset(),
-      model::offset(59));
-
-    // gc with timestamp +sep/2, should evict first segment
-    compact_and_prefix_truncate(
-      *disk_log, make_compaction_cfg(broker_t0 + (broker_ts_sep / 2)));
-    ASSERT_EQ(disk_log->segments().size(), 2);
-    ASSERT_EQ(
-      disk_log->segments().front()->offsets().get_base_offset(),
-      model::offset(10));
-    ASSERT_EQ(
-      disk_log->segments().back()->offsets().get_dirty_offset(),
-      model::offset(59));
-    // gc with timestamp +sep3/2, should evict another segment
-    compact_and_prefix_truncate(
-      *disk_log, make_compaction_cfg(broker_t0 + (3 * broker_ts_sep / 2)));
     ASSERT_EQ(disk_log->segments().size(), 1);
     ASSERT_EQ(
       disk_log->segments().front()->offsets().get_base_offset(),

tests/rptest/tests/retention_policy_test.py

@@ -863,3 +863,48 @@ def prefix_truncated():
         self.redpanda.wait_until(prefix_truncated,
                                  timeout_sec=30,
                                  backoff_sec=1)
+
+    @cluster(num_nodes=2)
+    def test_past_timestamps(self):
+        """
+        While future record timestamps should be adjusted back to the broker
+        timestamp to avoid blocking retention enforcement, record timestamps in the past
+        should still be respected.
+        """
+
+        # Set `retention.ms` to 23 hours
+        retention_ms = 23 * 3600 * 1000
+        self.client().alter_topic_config(self.topic, 'retention.ms',
+                                         retention_ms)
+
+        # A fictional artificial timestamp base in milliseconds (one day previous)
+        past_timestamp = (int(time.time()) - 24 * 3600) * 1000
+
+        # Produce a run of messages with CreateTime-style timestamps, each
+        # record having a timestamp 1ms greater than the last.
+        msg_size = 14000
+        segments_count = 10
+        msg_count = (self.segment_size // msg_size) * segments_count
+
+        # Write msg_count messages with timestamps in the past
+        producer = KgoVerifierProducer(
+            context=self.test_context,
+            redpanda=self.redpanda,
+            topic=self.topic,
+            msg_size=msg_size,
+            msg_count=(self.segment_size // msg_size) * segments_count,
+            fake_timestamp_ms=past_timestamp,
+            batch_max_bytes=msg_size * 2)
+        producer.start()
+        producer.wait()
+
+        def prefix_truncated():
+            segs = self.redpanda.node_storage(self.redpanda.nodes[0]).segments(
+                "kafka", self.topic, 0)
+            self.logger.debug(f"Segments: {segs}")
+            return len(segs) <= 1
+
+        # Expect to see prefix truncation of day old records with `retention.ms=23h`.
+        self.redpanda.wait_until(prefix_truncated,
+                                 timeout_sec=30,
+                                 backoff_sec=1)