krabka_log/log/verbatim.rs
1//! The zero-copy passthrough append: a producer batch written to the log
2//! byte-for-byte, with no re-encode. Replicated controls decode their marker
3//! for transaction bookkeeping; data batches use only the header.
4//!
5//! Only `base_offset` and `partition_leader_epoch` are patched into the
6//! bytes, and both sit outside the CRC region, so the producer's body and
7//! checksum reach disk exactly as they arrived.
8//!
9//! A `message.timestamp.type=LogAppendTime` partition is the one exception:
10//! it also rewrites the timestamp-type attribute bit, `max_timestamp`, and
11//! the CRC over them, which is exactly the three fields Kafka's
12//! `LogValidator` rewrites. The producer's records are still stored as they
13//! arrived.
14
15use {
16 super::{Log, append::AppendRoll, open::DataProducerAppend},
17 crate::error::LogError,
18 bytes::{Bytes, BytesMut},
19 krabka_ids::{LeaderEpoch, Offset, ProducerId},
20 krabka_protocol::records::{
21 Attributes, CRC_COVERAGE_START, HEADER_LEN, RecordBatch, RecordBatchHeader, TimestampType,
22 crc32c,
23 },
24 krabka_units::prelude::{ByteSize, ByteSizeExt as _},
25 std::ops::Range,
26 tracing::{field, instrument},
27};
28
29/// Byte range of the `crc` field in the v2 batch header: the four bytes that
30/// sit immediately before the region the CRC covers.
31const CRC_RANGE: Range<usize> = CRC_COVERAGE_START - 4..CRC_COVERAGE_START;
32
33/// Byte range of `attributes` in the v2 batch header. It is the first field
34/// the CRC covers, so it starts at [`CRC_COVERAGE_START`] and is two bytes
35/// wide.
36const ATTRIBUTES_RANGE: Range<usize> = CRC_COVERAGE_START..CRC_COVERAGE_START + 2;
37
38/// Byte range of `max_timestamp` in the v2 batch header: `attributes` (2) plus
39/// `last_offset_delta` (4) plus `base_timestamp` (8) after
40/// [`CRC_COVERAGE_START`], and eight bytes wide.
41const MAX_TIMESTAMP_RANGE: Range<usize> = CRC_COVERAGE_START + 14..CRC_COVERAGE_START + 22;
42
43/// A batch that the log appends **verbatim**, with no re-encode.
44///
45/// The produce zero-copy passthrough path uses this type. It carries the
46/// producer's exact wire bytes plus the header fields the log needs for
47/// offset assignment, LSO and transaction tracking, the leader-epoch
48/// checkpoint, and the time index. The caller has already read all of those
49/// fields from the batch header with a borrowed header-only decode.
50///
51/// The append patches only `base_offset` and `partition_leader_epoch` into a
52/// writable copy of [`Self::bytes`]. Both fields sit outside the CRC region.
53/// The log writes the body and the CRC byte-for-byte as the producer sent
54/// them.
55///
56/// Replicated control batches decode their inner marker for shared transaction
57/// bookkeeping while storing the original bytes. Data batches keep the
58/// header-only path; the producer passthrough caller rejects control batches.
59#[derive(Debug, Clone)]
60pub struct VerbatimBatch {
61 /// The producer's verbatim v2 batch bytes (CRC-validated by the caller).
62 pub bytes: Bytes,
63 /// `last_offset_delta` from the header: how many offsets the batch spans.
64 pub last_offset_delta: i32,
65 /// `max_timestamp` from the header (for `max_timestamp` + time index).
66 pub max_timestamp: i64,
67 /// Leader epoch to stamp into the batch (`partition_leader_epoch`).
68 pub leader_epoch: LeaderEpoch,
69 /// `producer_id` from the header (for LSO/transaction tracking).
70 pub producer_id: ProducerId,
71 /// `producer_epoch` from the header.
72 pub producer_epoch: i16,
73 /// `base_sequence` from the header.
74 pub base_sequence: i32,
75 /// `true` when the batch's attributes mark it transactional.
76 pub is_transactional: bool,
77}
78
79impl VerbatimBatch {
80 /// Preserve a validated batch's wire bytes and copy its header bookkeeping fields.
81 #[must_use]
82 pub fn from_header(header: &RecordBatchHeader, bytes: Bytes) -> Self {
83 Self {
84 bytes,
85 last_offset_delta: header.last_offset_delta.get(),
86 max_timestamp: header.max_timestamp.get(),
87 leader_epoch: LeaderEpoch(header.partition_leader_epoch.get()),
88 producer_id: ProducerId(header.producer_id.get()),
89 producer_epoch: header.producer_epoch.get(),
90 base_sequence: header.base_sequence.get(),
91 is_transactional: Attributes(header.attributes.get()).is_transactional(),
92 }
93 }
94
95 /// This batch with Kafka's log-append time stamped into it.
96 ///
97 /// This is `LogValidator`'s `batch.setMaxTimestamp(LOG_APPEND_TIME, now)`
98 /// applied to wire bytes rather than to a decoded batch: it moves the
99 /// timestamp-type attribute bit, `max_timestamp`, and the CRC that covers
100 /// them, and nothing else. `base_timestamp` and the per-record deltas stay
101 /// as the producer wrote them, exactly as in Kafka, because a reader
102 /// substitutes `max_timestamp` for every record while the bit is set.
103 ///
104 /// The three fields sit at fixed offsets in the header, so the patch needs
105 /// no decode. Two of them are inside the CRC region, which is why this is
106 /// the one place the passthrough path recomputes a CRC: over the patched
107 /// header tail and the producer's own body, which never changes. The batch
108 /// bytes are copied once, so only a `LogAppendTime` partition pays for the
109 /// copy and a `CreateTime` one keeps the zero-copy append whole.
110 ///
111 /// # Errors
112 ///
113 /// [`LogError::Corrupt`] when the batch is shorter than its header.
114 fn stamped_with_log_append_time(&self, now: i64) -> Result<Self, LogError> {
115 let short = || LogError::Corrupt("verbatim batch is shorter than its header".into());
116 let mut bytes = BytesMut::from(self.bytes.as_ref());
117 let attributes = bytes
118 .get(ATTRIBUTES_RANGE)
119 .and_then(|attributes| <[u8; 2]>::try_from(attributes).ok())
120 .ok_or_else(short)?;
121 let attributes = Attributes(i16::from_be_bytes(attributes))
122 .with_timestamp_type(TimestampType::LogAppendTime);
123 bytes
124 .get_mut(ATTRIBUTES_RANGE)
125 .ok_or_else(short)?
126 .copy_from_slice(&attributes.0.to_be_bytes());
127 bytes
128 .get_mut(MAX_TIMESTAMP_RANGE)
129 .ok_or_else(short)?
130 .copy_from_slice(&now.to_be_bytes());
131 // The CRC covers the header from `attributes` to the end of the batch,
132 // and the body follows the header without a gap, so one pass over the
133 // tail of the buffer is the whole covered region.
134 let crc = crc32c(bytes.get(CRC_COVERAGE_START..).ok_or_else(short)?);
135 bytes
136 .get_mut(CRC_RANGE)
137 .ok_or_else(short)?
138 .copy_from_slice(&crc.to_be_bytes());
139 Ok(Self {
140 bytes: bytes.freeze(),
141 max_timestamp: now,
142 ..self.clone()
143 })
144 }
145}
146
147impl Log {
148 /// Append a producer batch **verbatim** and return the assigned
149 /// `base_offset` plus the log-append stamp, as [`Log::append`] does.
150 ///
151 /// The log does not decode or re-encode the batch, and it takes
152 /// `base_offset` from the log's current end. This is the produce
153 /// zero-copy passthrough path. The caller has already CRC-validated the
154 /// bytes and read the header fields into [`VerbatimBatch`]. The log
155 /// patches `base_offset` and `partition_leader_epoch`, which both sit
156 /// outside the CRC region, then writes the bytes as they are. Offset
157 /// assignment, segment roll, flush, LSO and transaction tracking, and the
158 /// leader-epoch checkpoint behave exactly as in [`Log::append`]. The
159 /// verbatim path and the owned path differ only in how the log gets the
160 /// batch bytes, not in any log-level invariant.
161 #[instrument(
162 level = "debug",
163 skip_all,
164 fields(
165 assigned_base = field::Empty,
166 leader_epoch = batch.leader_epoch.0,
167 bytes = batch.bytes.len(),
168 ),
169 err,
170 )]
171 /// # Errors
172 /// Returns an error when log I/O fails, a record or index is corrupt, or the requested offset violates the segment state.
173 pub fn append_verbatim(
174 &mut self,
175 batch: &VerbatimBatch,
176 ) -> Result<(Offset, Option<i64>), LogError> {
177 // A fenced log refuses every append with `Fenced`, before any of
178 // the append's own validation can report a different error.
179 self.ensure_unfenced()?;
180 // KFC-1, before an offset is assigned and before anything is written,
181 // so a refused batch leaves the log exactly as it found it.
182 self.reject_backwards_schedule(batch.max_timestamp)?;
183 let leader_epoch = batch.leader_epoch;
184 let assigned_base = self.append_at_expected_offset();
185 tracing::Span::current().record("assigned_base", assigned_base.0);
186 // A batch too short to hold a header cannot be patched. Passing it on
187 // unchanged lets the segment append report the truncation it already
188 // reports, rather than turning it into a panic here.
189 let stamp = self
190 .log_append_time_stamp()
191 .filter(|_| batch.bytes.len() >= HEADER_LEN);
192 let stamped = stamp
193 .map(|now| batch.stamped_with_log_append_time(now))
194 .transpose()?;
195 let batch = stamped.as_ref().unwrap_or(batch);
196 self.append_verbatim_preserving_offset(batch, assigned_base)?;
197 self.assign_appended_epoch(leader_epoch, assigned_base)?;
198 Ok((assigned_base, stamp))
199 }
200
201 /// Append a producer batch **verbatim** at a caller-supplied base offset.
202 ///
203 /// `base_offset` must not be below the log's current
204 /// [`Log::log_end_offset`]. If it is, this method returns
205 /// [`LogError::OffsetMismatch`] and appends nothing. A `base_offset` above
206 /// the end is a hole in the offsets, as a compacted leader's log has, and
207 /// is appended as it is. On success the log stamps the stored batch with
208 /// `base_offset` and the batch's leader epoch. Control batches decode the
209 /// marker for shared bookkeeping; CRC-covered bytes are never re-encoded.
210 ///
211 /// # Errors
212 /// Returns [`LogError::OffsetMismatch`] when `base_offset` is below the
213 /// log end offset. It also propagates segment and checkpoint I/O errors
214 /// and validation errors.
215 #[instrument(
216 level = "debug",
217 skip_all,
218 fields(
219 supplied_base = base_offset.0,
220 leader_epoch = batch.leader_epoch.0,
221 bytes = batch.bytes.len(),
222 ),
223 err,
224 )]
225 pub fn append_verbatim_at(
226 &mut self,
227 batch: &VerbatimBatch,
228 base_offset: Offset,
229 ) -> Result<Offset, LogError> {
230 // A fenced log refuses every append with `Fenced`, before any of
231 // the append's own validation can report a different error.
232 self.ensure_unfenced()?;
233 self.admit_append_at(base_offset)?;
234
235 let leader_epoch = batch.leader_epoch;
236 let attributes = batch
237 .bytes
238 .get(ATTRIBUTES_RANGE)
239 .and_then(|attributes| <[u8; 2]>::try_from(attributes).ok())
240 .ok_or_else(|| LogError::Corrupt("verbatim batch is missing its attributes".into()))?;
241 if Attributes(i16::from_be_bytes(attributes)).is_control_batch() {
242 let mut wire = batch.bytes.as_ref();
243 let mut decoded = RecordBatch::decode(&mut wire)?;
244 if !wire.is_empty() {
245 return Err(LogError::Corrupt(
246 "verbatim control append contains trailing bytes".into(),
247 ));
248 }
249 decoded.base_offset = base_offset.0;
250 decoded.partition_leader_epoch = leader_epoch.0;
251 self.append_preserving_offset_inner(&decoded, None, Some(&batch.bytes))?;
252 } else {
253 self.append_verbatim_preserving_offset(batch, base_offset)?;
254 }
255 self.assign_appended_epoch(leader_epoch, base_offset)?;
256 Ok(base_offset)
257 }
258
259 /// Verbatim counterpart of [`Log::append_preserving_offset`].
260 ///
261 /// This function rolls the segment if necessary, appends the verbatim
262 /// bytes to the active segment, honors `flush_on_append`, and updates the
263 /// LSO from the batch's transactional and producer metadata. It mirrors
264 /// the non-control branches of the owned path. Control batches never
265 /// reach here, because they take the owned path.
266 fn append_verbatim_preserving_offset(
267 &mut self,
268 batch: &VerbatimBatch,
269 base_offset: Offset,
270 ) -> Result<(), LogError> {
271 self.ensure_unfenced()?;
272 self.rollover_flusher.check()?;
273 let Some((last_offset, _)) = krabka_verified::local_append_coordinates(
274 self.append_at_expected_offset().0,
275 base_offset.0,
276 batch.last_offset_delta,
277 ) else {
278 return Err(LogError::InvalidArgument(
279 "batch does not form a valid interval at the append frontier".into(),
280 ));
281 };
282 let last_offset = Offset(last_offset);
283 Self::data_producer_tail(
284 batch.producer_id,
285 batch.base_sequence,
286 batch.last_offset_delta,
287 base_offset,
288 )?;
289 // The follower uses the same append-time policy and roll check as the
290 // owned path, against this replicated batch's bytes and timestamp.
291 let AppendRoll {
292 index_interval,
293 flush_on_append,
294 } = self.roll_for_append(
295 || ByteSize::from_bytes(u64::try_from(batch.bytes.len()).unwrap_or(u64::MAX)),
296 batch.max_timestamp,
297 )?;
298
299 let result = (|| {
300 let active = self.active.as_mut().ok_or_else(super::no_active_segment)?;
301 active.append_verbatim(
302 &batch.bytes,
303 base_offset,
304 batch.last_offset_delta,
305 batch.max_timestamp,
306 batch.leader_epoch,
307 index_interval,
308 )?;
309
310 let is_transactional = batch.is_transactional && batch.producer_id.get() >= 0;
311 self.flush_after_append(
312 batch.last_offset_delta,
313 flush_on_append || (self.stamp_source.is_some() && !is_transactional),
314 )?;
315 #[cfg(not(target_os = "wasi"))]
316 self.pace_writeback(batch.bytes.len())?;
317
318 if !is_transactional {
319 self.record_stamp(base_offset, last_offset)?;
320 }
321
322 let pid = batch.producer_id;
323 if is_transactional {
324 self.record_pending_transaction(pid, base_offset, last_offset);
325 }
326 self.refresh_lso()?;
327
328 self.update_data_producer_entry(DataProducerAppend {
329 producer_id: batch.producer_id,
330 producer_epoch: batch.producer_epoch,
331 base_sequence: batch.base_sequence,
332 last_offset_delta: batch.last_offset_delta,
333 base_offset,
334 timestamp: batch.max_timestamp,
335 is_transactional,
336 })?;
337 if is_transactional {
338 self.clear_verification_after_append(pid, batch.producer_epoch, false);
339 }
340 Ok(())
341 })();
342
343 self.complete_append_state(result, base_offset)
344 }
345}
346
347#[cfg(test)]
348mod tests;