perf(server-ng): compute batch checksum in a single produce pass

core/partitions/src/iggy_partition.rs

@@ -57,7 +57,8 @@ use server_common::{
     MESSAGE_ALIGN, Message, SegmentStorage,
     iobuf::{Frozen, Owned},
     send_messages2::{
-        convert_request_message, decode_prepare_slice, stamp_prepare_for_persistence,
+        ChecksumMode, convert_request_message, decode_prepare_slice, decode_prepare_slice_trusted,
+        stamp_prepare_for_persistence, verify_received_send_messages,
     },
     sharding::IggyNamespace,
 };
@@ -1149,7 +1150,13 @@ where
             );
 
             let message = if message.header().operation == Operation::SendMessages {
-                match convert_request_message(namespace, message) {
+                // Skip the batch-checksum pass: on the partition ingest path
+                // nothing reads it before `stamp_prepare_for_persistence`
+                // recomputes it over the stamped header. An already-canonical
+                // batch (native v2, or the plane's pre-encrypt convert output)
+                // returns early above, so Skip only affects the legacy
+                // transcode, whose output goes straight to project/stamp.
+                match convert_request_message(namespace, message, ChecksumMode::Skip) {
                     Ok(message) => message,
                     Err(error) => {
                         emit_partition_diag(
@@ -1532,6 +1539,32 @@ where
                 );
             }
         }
+        // First blob-integrity check on the replicated path. The consensus
+        // layer never validates the body (PrepareHeader integrity fields are
+        // inert zeros) and the batch checksum is recomputed locally at stamp,
+        // so a follower must verify each message's stamp-invariant per-message
+        // checksum before journaling transit bytes. Follower-only: the primary
+        // (and single-node self-replicate) produced these bytes and already
+        // checked the client batch at ingest, so they must not pay this pass.
+        // Fail closed on mismatch - drop without journaling, forwarding, or
+        // acking; the primary retransmits on prepare-timeout.
+        if is_backup
+            && header.operation == Operation::SendMessages
+            && let Err(error) = verify_received_send_messages(message.as_slice())
+        {
+            emit_partition_diag(
+                tracing::Level::WARN,
+                &PartitionDiagEvent::new(
+                    self.diag_ctx(),
+                    "rejecting replicated send_messages: per-message checksum mismatch",
+                )
+                .with_operation(header.operation)
+                .with_op(header.op)
+                .with_error(error.to_string()),
+            );
+            return;
+        }
+
         // Durability-before-ack: clone for chain-replicate, forward only
         // AFTER apply_replicated_operation persists. Forward-first would
         // give downstream an op whose WAL entry we never wrote, that violates
@@ -1941,11 +1974,15 @@ where
                         }
                         continue;
                     }
-                    // A resident committed SendMessages entry decoded once at append
-                    // (the offset index) with its checksum stamped over these exact
-                    // bytes, so it must decode again here. Guard the invariant for a
-                    // future disk read-back path that could make decode fallible.
-                    let Ok(batch) = decode_prepare_slice(entry.as_slice()) else {
+                    // Resident committed SendMessages entry: this node stamped it
+                    // in `append_messages` (recomputing the batch checksum over these
+                    // exact bytes), so a validating re-decode would only re-hash ~1
+                    // MiB to confirm our own write. Trust the structural decode; the
+                    // batch-checksum recompute belongs at network ingress (repair
+                    // validation + the follower receive gate), not on locally-stamped
+                    // bytes. Guard the invariant for a future disk read-back path that
+                    // could make decode fallible.
+                    let Ok(batch) = decode_prepare_slice_trusted(entry.as_slice()) else {
                         tracing::error!(
                             target: "iggy.partitions.diag",
                             namespace_raw = self.namespace().inner(),
@@ -2345,8 +2382,11 @@ where
         let Some(entry) = self.log.journal().inner.entry(prepare_header).await else {
             return Err(IggyError::InvalidCommand);
         };
-        let batch =
-            decode_prepare_slice(entry.as_slice()).map_err(|_| IggyError::InvalidCommand)?;
+        // Trusted (no batch-hash): the entry was read back from this replica's
+        // own journal, where it was stamped/validated at append; only header
+        // stats are needed, so re-hashing the ~1 MiB blob is redundant.
+        let batch = decode_prepare_slice_trusted(entry.as_slice())
+            .map_err(|_| IggyError::InvalidCommand)?;
         let message_count = batch.message_count();
         if message_count == 0 {
             return Ok(None);

core/partitions/src/iggy_partitions.rs

@@ -26,7 +26,7 @@ use iggy_binary_protocol::{
     Command2, ConsensusHeader, Operation, PrepareHeader, PrepareOkHeader, RequestHeader,
 };
 use message_bus::MessageBus;
-use server_common::send_messages2::{convert_request_message, encrypt_batch_request};
+use server_common::send_messages2::{ChecksumMode, convert_request_message, encrypt_batch_request};
 use server_common::sharding::{IggyNamespace, LocalIdx, ShardId};
 #[cfg(debug_assertions)]
 use std::cell::Cell;
@@ -512,7 +512,11 @@ where
         let message = if message.header().operation == Operation::SendMessages
             && let Some(encryptor) = &self.config().encryptor
         {
-            let canonical = convert_request_message(namespace, message)
+            // Compute the batch checksum: this canonical output is validated by
+            // `encrypt_batch_request`'s decode before re-encryption, and the
+            // re-encrypted batch (checksum kept by `encrypt_batch_request`) then
+            // re-enters `convert` as the canonical-vs-legacy discriminator.
+            let canonical = convert_request_message(namespace, message, ChecksumMode::Compute)
                 .and_then(|message| encrypt_batch_request(message, encryptor));
             match canonical {
                 Ok(message) => message,

core/partitions/src/journal.rs

@@ -19,7 +19,7 @@ use iggy_binary_protocol::{Operation, PrepareHeader};
 use journal::{Journal, Storage};
 use server_common::{
     iobuf::{Frozen, Owned},
-    send_messages2::{COMMAND_HEADER_SIZE, SendMessages2Ref, decode_prepare_slice},
+    send_messages2::{COMMAND_HEADER_SIZE, SendMessages2Ref, decode_prepare_slice_trusted},
 };
 use std::io;
 use std::{
@@ -538,8 +538,12 @@ impl PartitionJournal<PartitionJournalMemStorage> {
         // One decode feeds both the offset/timestamp index (keyed on
         // `origin_timestamp`) and the surfaced accounting meta (`base_timestamp`,
         // size, count); the two timestamps are distinct fields, do not conflate.
+        // Trusted (no batch-hash): every entry reaching append was just stamped
+        // by `stamp_prepare_for_persistence` (its checksum recomputed over this
+        // exact blob) or re-appended from an already-validated resident entry,
+        // so re-hashing the ~1 MiB blob here only to read the header is waste.
         let (index_offset_timestamp, meta) = if header.operation == Operation::SendMessages {
-            match decode_prepare_slice(entry.as_slice()) {
+            match decode_prepare_slice_trusted(entry.as_slice()) {
                 Ok(batch) if batch.message_count() != 0 => {
                     let message_count = batch.message_count();
                     let meta = RetainedBatchMeta {
@@ -885,7 +889,10 @@ fn try_push_resident_entry(
     if header.operation != Operation::SendMessages {
         return;
     }
-    let Ok(batch) = decode_prepare_slice(prepare.as_slice()) else {
+    // Resident entries were locally stamped in `append_messages` or validated
+    // at repair ingress, so a validating re-decode would only re-hash our own
+    // write. See the invariant note at the committed-prefix flush walk.
+    let Ok(batch) = decode_prepare_slice_trusted(prepare.as_slice()) else {
         return;
     };
     let Some(selection) = select_batch_slice(&batch, query, *matched_messages) else {

core/server-ng/src/responses.rs

@@ -1437,7 +1437,7 @@ where
 /// Size of the in-storage (`IggyMessage2`) per-message header inside a
 /// `SendMessages2` batch blob: `checksum`(8) + `id`(16) + `offset_delta`(4)
 /// + `timestamp_delta`(4) + `user_headers_length`(4) + `payload_length`(4)
-/// + reserved(8). See `server_common::send_messages2::from_legacy_request`.
+/// + reserved(8). See `server_common::send_messages2::SendMessages2Owned::from_messages`.
 const STORED_MESSAGE_HEADER_SIZE: usize = 48;
 
 /// Build the `PolledMessages` reply body from the owning shard's poll