storage,kafka: extra debug logs for timestamps

Hunting redpanda-data#3924

src/v/kafka/server/handlers/produce.cc

@@ -239,6 +239,26 @@ static partition_produce_stages produce_topic_partition(
           model::timestamp_type::append_time, model::timestamp::now());
     }
 
+#ifndef NDEBUG
+    auto now = model::timestamp::now();
+    vlog(
+      klog.trace,
+      "produce batch timestamps: now={} first {} max {}",
+      now,
+      batch.header().first_timestamp,
+      batch.header().max_timestamp);
+    if (
+      batch.header().first_timestamp > now
+      || batch.header().max_timestamp > now) {
+        vlog(
+          klog.warn,
+          "produce batch timestamps in future: now={} first {} max {}",
+          now,
+          batch.header().first_timestamp,
+          batch.header().max_timestamp);
+    }
+#endif
+
     const auto& hdr = batch.header();
     auto bid = model::batch_identity::from(hdr);
     auto batch_size = batch.size_bytes();

src/v/storage/index_state.cc

@@ -59,6 +59,24 @@ bool index_state::maybe_index(
     // last one
     last_timestamp = std::max(first_timestamp, last_timestamp);
     max_timestamp = std::max(max_timestamp, last_timestamp);
+
+#ifndef NDEBUG
+    // If future-timestamped content creeps into our storage layer,
+    // it can disrupt time-based compaction: log when this happens.
+    // Related: https://github.com/redpanda-data/redpanda/issues/3924
+    auto now = model::timestamp::now();
+    if (max_timestamp > model::timestamp::now()) {
+        vlog(
+          stlog.warn,
+          "Updating index with max timestamp {} in future (now={}, "
+          "first_timestamp={}, base_timestamp={})",
+          max_timestamp,
+          now,
+          first_timestamp,
+          base_timestamp());
+    }
+#endif
+
     // always saving the first batch simplifies a lot of book keeping
     if (accumulator >= step || retval) {
         // We know that a segment cannot be > 4GB

src/v/storage/segment_index.cc

@@ -14,6 +14,7 @@
 #include "storage/index_state.h"
 #include "storage/logger.h"
 #include "vassert.h"
+#include "vlog.h"
 
 #include <seastar/core/coroutine.hh>
 #include <seastar/core/fstream.hh>
@@ -71,7 +72,12 @@ void segment_index::maybe_track(
           hdr.max_timestamp)) {
         _acc = 0;
     }
-    _needs_persistence = true;
+
+    vlog(
+      stlog.trace,
+      "maybe_track timestamps: base {} max {}",
+      base_timestamp(),
+      max_timestamp());
 }
 
 std::optional<segment_index::entry>