From 4be7193186fa786a5e70f3444c905018c2b48254 Mon Sep 17 00:00:00 2001 From: Andrew Wong Date: Wed, 29 Jul 2026 15:50:17 -0700 Subject: [PATCH] ct/l0: preserve the produce-time record crc on the read path Previously, when L0 materialized record batches from the payloads in object storage, we were resetting the record-batch crc and rederiving it from the materialized payload data, rather than using the crc in object storage that were written at produce time. This meant that we wouldn't have detected a corrupted bit at rest when served from object storage. Updates the record-batch materialization code to use the record-batch crc and only recompute the header_crc (needed since neither the placeholder nor the object storage batch have the full materialized-batch header_crc with correct offset and size). --- .../level_zero/reader/materialized_extent.cc | 9 +-- .../level_zero/stm/placeholder.cc | 14 ++++- src/v/cloud_topics/tests/BUILD | 1 + src/v/cloud_topics/tests/end_to_end_test.cc | 57 +++++++++++++++++++ 4 files changed, 71 insertions(+), 10 deletions(-) diff --git a/src/v/cloud_topics/level_zero/reader/materialized_extent.cc b/src/v/cloud_topics/level_zero/reader/materialized_extent.cc index 66cdae57053b3..1a004d5b79171 100644 --- a/src/v/cloud_topics/level_zero/reader/materialized_extent.cc +++ b/src/v/cloud_topics/level_zero/reader/materialized_extent.cc @@ -100,14 +100,9 @@ model::record_batch make_raft_data_batch(materialized_extent ext) { size() - model::packed_record_batch_header_size); auto header = storage::batch_header_from_disk_iobuf( std::move(header_bytes)); - // NOTE: the serialized raft_data batch doesn't have the offset set - // so we need to populate it from the placeholder batch. We also need - // to make sure that crc is correct. + + // base_offset comes from the placeholder; the serialized batch has none. header.base_offset = kafka::offset_cast(ext.meta.base_offset); - header.crc = model::crc_record_batch(header, records_bytes); - crc::crc32c crc; - model::crc_record_batch_header(crc, header); - header.header_crc = crc.value(); model::record_batch batch( header, std::move(records_bytes), diff --git a/src/v/cloud_topics/level_zero/stm/placeholder.cc b/src/v/cloud_topics/level_zero/stm/placeholder.cc index 87e910d242813..471644f6a7c28 100644 --- a/src/v/cloud_topics/level_zero/stm/placeholder.cc +++ b/src/v/cloud_topics/level_zero/stm/placeholder.cc @@ -9,7 +9,9 @@ */ #include "cloud_topics/level_zero/stm/placeholder.h" +#include "model/record.h" #include "model/record_batch_types.h" +#include "model/record_utils.h" #include "storage/record_batch_builder.h" namespace cloud_topics { @@ -79,13 +81,15 @@ ctp_placeholder parse_placeholder_batch(model::record_batch batch) { model::record_batch apply_placeholder_to_batch( const model::record_batch_header& placeholder_batch_header, model::record_batch uploaded_batch) { - // crcs and sizes are set later in reset_size_checksum_metadata model::record_batch_header merged_header{ .header_crc = 0, .size_bytes = 0, .base_offset = placeholder_batch_header.base_offset, .type = uploaded_batch.header().type, - .crc = 0, + // The produce-time value, which has to reach the consumer unchanged so + // that corruption in the records stays detectable. Every field it covers + // is reproduced below from the same batch it was computed over. + .crc = uploaded_batch.header().crc, // We need to use the same attributes for compression and timestamp types // which are changed in the placeholder batch .attrs = uploaded_batch.header().attrs, @@ -102,7 +106,11 @@ model::record_batch apply_placeholder_to_batch( .record_count = placeholder_batch_header.record_count, .ctx = placeholder_batch_header.ctx, }; - merged_header.reset_size_checksum_metadata(uploaded_batch.data()); + // size_bytes and header_crc describe the merged header, so both are + // derived here; .crc is the produce-time value set above and stands. + merged_header.size_bytes = model::packed_record_batch_header_size + + uploaded_batch.data().size_bytes(); + merged_header.header_crc = model::internal_header_only_crc(merged_header); return model::record_batch( merged_header, std::move(uploaded_batch).release_data(), diff --git a/src/v/cloud_topics/tests/BUILD b/src/v/cloud_topics/tests/BUILD index e4d3db7f408e1..ea63b49f2b9c3 100644 --- a/src/v/cloud_topics/tests/BUILD +++ b/src/v/cloud_topics/tests/BUILD @@ -40,6 +40,7 @@ redpanda_cc_gtest( "//src/v/cluster:self_test", "//src/v/cluster:topic_metrics_watcher", "//src/v/config", + "//src/v/container:chunked_vector", "//src/v/kafka/server/tests:kafka_test_utils", "//src/v/model", "//src/v/model:batch_builder", diff --git a/src/v/cloud_topics/tests/end_to_end_test.cc b/src/v/cloud_topics/tests/end_to_end_test.cc index 5e6cd5eacf38a..51fab613d820b 100644 --- a/src/v/cloud_topics/tests/end_to_end_test.cc +++ b/src/v/cloud_topics/tests/end_to_end_test.cc @@ -10,6 +10,7 @@ #include "cloud_io/tests/s3_imposter.h" #include "cloud_topics/level_zero/stm/ctp_stm.h" +#include "container/chunked_vector.h" #include "kafka/server/tests/list_offsets_utils.h" #include "kafka/server/tests/produce_consume_utils.h" #include "model/batch_builder.h" @@ -139,6 +140,62 @@ TEST_F(e2e_fixture, test_l0_path) { } } +// Regression test for an incorrect integity check: previous versions of +// Redpanda wouldn't detect corrupted data on the L0 path, and would instead +// re-CRC based on what was in the cloud. Guard against this by flipping a bit +// in an uploaded object and ensure the client doesn't see it (our client +// silently drops it). +TEST_F(e2e_fixture, test_corrupt_l0_object_fails_client_crc_check) { + // Disable reconciliation and disable the batch cache to ensure we read + // from L0 objects. + test_local_cfg.get("cloud_topics_disable_reconciliation_loop") + .set_value(true); + test_local_cfg.get("disable_batch_cache").set_value(true); + + const ss::sstring marker(64, 'A'); + auto* producer = make_producer(); + producer + ->produce_to_partition( + topic_name, model::partition_id(0), std::vector{{"key", marker}}) + .get(); + + auto puts = get_requests( + [&marker](const http_test_utils::request_info& req) { + return req.method == "PUT" + && req.content.find(marker) != ss::sstring::npos; + }); + ASSERT_EQ(puts.size(), 1); + const auto l0_url = puts.front().url; + auto corrupted = puts.front().content; + auto flip_at = corrupted.find(marker) + marker.size() / 2; + corrupted[flip_at] = static_cast(corrupted[flip_at] ^ 0x01); + + // Republish the object with the corrupted body. + const auto key = l0_url.substr(1); + remove_expectations(chunked_vector::single(key)); + add_expectations( + chunked_vector::single( + expectation{.url = key, .body = corrupted})); + + auto* consumer = make_consumer(); + auto records = consumer + ->raw_consume_from_partition( + topic_name, model::partition_id(0), model::offset(0)) + .get(); + + // Sanity check that we actually got the object from storage. + auto gets = get_requests( + [&l0_url](const http_test_utils::request_info& req) { + return req.method == "GET" && req.url == l0_url; + }); + ASSERT_FALSE(gets.empty()) << "fetch never downloaded the L0 object"; + + EXPECT_TRUE(records.empty()) + << "client accepted " << records.size() + << " record(s) from a corrupted L0 object: the read path replaced the " + "produce-time record crc with one computed over the corrupted bytes"; +} + TEST_F(e2e_fixture, timequery) { lconf().log_message_timestamp_after_max_ms.set_value( serde::max_serializable_ms);