storage: fix compaction minting future record timestamps from strictly-past input

[Aangbaeck]

When compaction drops some of a batch's records, the batch is re-encoded.
The old code advanced the new first_timestamp to the first surviving
record but left the per-record timestamp deltas unchanged (still relative to
the batch's original first_timestamp). On read, each record then resolves
to new_first_timestamp + old_delta, i.e. shifted forward by the dropped
base record's delta. With a tombstone dropping the base of an out-of-order
batch, that shift is large and positive, so strictly-past records read back
with timestamps months in the future.

Fix: re-base the surviving deltas onto the new first_timestamp, the same way
Kafka does in MemoryRecordsBuilder.appendWithOffset. Added
record::set_timestamp_delta() to keep _size_bytes in sync with the
re-encoded vint width. Applied to both compaction paths
(copy_data_segment_reducer::filter and compaction::filter::do_filter_batch).

Regression tests build an out-of-order batch whose delta-0 base record is
compacted away, run compaction, and assert record timestamps are unchanged and
max_timestamp equals the greatest surviving timestamp. They fail on the old
reducer and pass with the fix (compaction/tests/reducer_test.cc for the
do_filter_batch path, storage/tests/compaction_e2e_test.cc for the
sliding-window path).

Notes:

• max_timestamp is the new first_timestamp plus the greatest surviving
  re-based delta, i.e. the greatest surviving absolute timestamp. (The previous
  code tracked the last delta, which is wrong for out-of-order batches.)
• Offsets are intentionally not re-based: base_offset is preserved as before,
  matching the DefaultRecordBatch producer-state semantics already documented in
  the reducer.

[Copilot]

Pull request overview

This PR fixes a log-compaction re-encode bug where partially-compacted batches could end up with a corrupted (shifted) batch base timestamp, causing records that were produced with strictly-past timestamps to be read back with future timestamps after compaction. The fix keeps per-record timestamp deltas unchanged while preserving the original batch first_timestamp, and recomputes max_timestamp using the maximum surviving timestamp delta (rather than assuming record iteration order matches timestamp order).

Changes:

• Track the maximum surviving timestamp_delta during batch filtering (rather than the first/last surviving deltas).
• Preserve the original record_batch_header.first_timestamp during compaction re-encode to keep unchanged deltas semantically valid.
• Recompute max_timestamp from first_timestamp + max_surviving_timestamp_delta for CREATE_TIME batches.

[CLAassistant]

All committers have signed the CLA.

[WillemKauf]

Thanks for the PR and raising the issue - though I wish the cover letter was hand written and not paragraphs of AI (the issue here can be summarized in a few sentences at most).

I also have some thoughts on how we should be doing the right thing (e.g. mirroring Apache Kafka compaction behavior) here.

[WillemKauf]

Also, we'll need to apply the logic of this fix to compaction::filter::do_filter_batch() as well.

redpanda/src/v/compaction/filter.cc

Lines 76 to 101 in 29dece6

	co_await b.for_each_record_async([&rec_count,
	&first_timestamp_delta,
	&last_timestamp_delta,
	&ret,
	&keep_idx,
	&offset_deltas](model::record record) {
	// contains the key
	if (
	keep_idx < offset_deltas.size()
	&& offset_deltas[keep_idx] == record.offset_delta()) {
	++keep_idx;
	/*
	* TODO when we further optimize lazy record materialization ot
	* make use of views we can avoid this re-encoding by copying or
	* sharing the view. either way, we were building
	* record batch with the uncompressed records so they were being
	* re-encoded.
	*/
	if (!first_timestamp_delta) {
	first_timestamp_delta = record.timestamp_delta();
	}
	last_timestamp_delta = record.timestamp_delta();
	model::append_record_to_buffer(ret, record);
	++rec_count;
	}
	});

[WillemKauf]

Commits like b9090c2 also shouldn't be in the PR

[Aangbaeck]

Reworked per the review — thanks for the detailed feedback.

• Re-based the surviving records' timestamp deltas onto the new first_timestamp (matching MemoryRecordsBuilder.appendWithOffset) instead of preserving the old one, and added record::set_timestamp_delta() to keep the encoded vint width / _size_bytes in sync.
• Applied the same fix to compaction::filter::do_filter_batch().
• Condensed the regression test into compaction_e2e_test.cc (dropped the separate file) and added a matching do_filter_batch test in compaction/tests/reducer_test.cc.
• Trimmed the comments; the incident detail now lives in the commit message.
• Rebased to a single commit — the merge commit is gone.

Verified against the base: both tests fail on the old reducer and pass with the fix (a strictly-past record reads back ~30 days in the future without it).

[WillemKauf]

Reworked per the review — thanks for the detailed feedback.

Thank you for the updates, though I'd prefer to feel like I'm talking to a human here 🙂

Changes LGTM, looking for one more added test and a slight comment update.

You'll need to run clang-format on the changes to pass our linter checks- we have a tools/format-cc.sh in the tree for this.

[WillemKauf]

Can we get a test for this specifically? Something as simple as calling set_timestamp_delta() on an existing record and asserting that the _size_bytes is accurate. model/tests/record_batch_test.cc already has examples of this

[Aangbaeck]

Added SetTimestampDeltaKeepsSizeBytesAccurate to record_batch_test.cc. It sets deltas that cross the vint width boundaries (63/64, -64/-65, ±14 days in ms) and re-checks the serialized size with check_serialization_size after each one.

[WillemKauf]

Used by compaction when the batch's base record is dropped and first_timestamp advances.

I don't like narrow comments that describe where a function is used, as it can quickly become out of date.

[Aangbaeck]

Fixed!

[Aangbaeck]

Finally, some meta - yes there is a real person behind this. 😉 But I just want to fix stuff.

Have a nice summer! 🌞