1#![allow(dead_code)]
233
234use crate::io::{FileSyncType, SharedBufferData};
235use crate::sync::Arc;
236use crate::sync::RwLock;
237use crate::turso_assert;
238use crate::{
239 alloc::{ConcurrentAllocator, TursoAllocator},
240 io::{CompletionGroup, ReadComplete},
241 io_yield_one,
242 mvcc::database::{LogRecord, MVTableId, Row, RowID, RowKey, RowVersion, SortableIndexKey},
243 return_if_io,
244 storage::sqlite3_ondisk::{
245 read_varint, read_varint_partial, varint_len, write_varint_to_vec, DatabaseHeader,
246 },
247 types::{IOCompletions, IOResult, IndexInfo},
248 util::IOExt as _,
249 Buffer, Completion, CompletionError, LimboError, Result,
250};
251
252use crate::storage::encryption::EncryptionContext;
253use crate::File;
254
255pub const DEFAULT_LOG_CHECKPOINT_THRESHOLD: i64 = 4120 * 1000;
258
259pub type OnSerializationComplete<'a> =
262 Option<&'a dyn Fn(SharedBufferData, u32) -> crate::Result<()>>;
263
264const LOG_MAGIC: u32 = 0x4C4D4C32; const LOG_VERSION_V2: u8 = 2;
266const LOG_VERSION: u8 = 3;
267pub const LOG_HDR_SIZE: usize = 56;
268const LOG_HDR_SALT_START: usize = 8;
269const LOG_HDR_SALT_SIZE: usize = 8;
270const LOG_HDR_RESERVED_START: usize = LOG_HDR_SALT_START + LOG_HDR_SALT_SIZE; const LOG_HDR_CRC_START: usize = 52;
272const LOG_HDR_RESERVED_SIZE: usize = LOG_HDR_CRC_START - LOG_HDR_RESERVED_START; pub(crate) const FRAME_MAGIC: u32 = 0x5854564D; pub(crate) const EXT_FRAME_MAGIC: u32 = 0x5845564D; const END_MAGIC: u32 = 0x4554564D; pub(crate) const ENCRYPTED_PAYLOAD_CHUNK_SIZE: usize = 32 * 1024;
279const ENCRYPTED_CHUNK_AAD_SIZE: usize = 32;
282
283const OP_UPSERT_TABLE: u8 = 0;
284const OP_DELETE_TABLE: u8 = 1;
285const OP_UPSERT_INDEX: u8 = 2;
286const OP_DELETE_INDEX: u8 = 3;
287const OP_UPDATE_HEADER: u8 = 4;
289
290const OP_FLAG_BTREE_RESIDENT: u8 = 1 << 0;
291const OP_FLAG_PORTABLE_EXTENSION: u8 = 1 << 1;
292const OP_ALLOWED_FLAGS: u8 = OP_FLAG_BTREE_RESIDENT | OP_FLAG_PORTABLE_EXTENSION;
293const OP_EXT_FIELD_DELETE_IDENTITY_RECORD: u64 = 1;
294const OP_EXT_FIELD_DELETE_PK_RECORD: u64 = 2;
295const OP_EXT_FIELD_DELETE_ROWID: u64 = 3;
296
297#[derive(Default)]
298struct DeletePortableExtension {
299 identity_record: Vec<u8>,
300 pk_record: Vec<u8>,
301}
302
303const TX_HEADER_SIZE_V2: usize = 24; const TX_HEADER_SIZE: usize = TX_HEADER_SIZE_V2;
305pub(crate) const TX_EXT_HEADER_SIZE: usize =
309 TX_HEADER_SIZE + 8 + 4 + 4 ;
310const TX_TRAILER_SIZE: usize = 8; const TX_MIN_FRAME_SIZE_V2: usize = TX_HEADER_SIZE_V2 + TX_TRAILER_SIZE; const TX_MIN_FRAME_SIZE: usize = TX_HEADER_SIZE + TX_TRAILER_SIZE; const TX_FRAME_FLAG_HAS_EXTENSION_BLOCK: u32 = 1 << 0;
314const EXTENSION_RECORD_HEADER_SIZE: usize = 8; const EXTENSION_TYPE_PORTABLE_CHANGES: u16 = 1;
316
317pub(crate) const LOG_RECORD_PREFIX_SIZE: usize = LOG_HDR_SIZE + TX_HEADER_SIZE;
319
320fn encrypted_payload_chunk_count(payload_size: usize, chunk_size: usize) -> usize {
321 if payload_size == 0 {
322 0
323 } else {
324 payload_size.div_ceil(chunk_size)
325 }
326}
327
328fn encrypted_chunk_plaintext_len(
333 payload_size: usize,
334 chunk_index: usize,
335 chunk_size: usize,
336) -> Result<usize> {
337 let chunk_start = chunk_index.checked_mul(chunk_size).ok_or_else(|| {
338 LimboError::Corrupt(format!(
339 "encrypted chunk offset overflow: chunk_index={chunk_index}, chunk_size={chunk_size}"
340 ))
341 })?;
342 if chunk_start >= payload_size {
343 return Err(LimboError::Corrupt(format!(
344 "encrypted chunk index {chunk_index} out of range for payload_size={payload_size}"
345 )));
346 }
347 Ok((payload_size - chunk_start).min(chunk_size))
348}
349
350fn encrypted_chunk_blob_size(
352 plaintext_len: usize,
353 tag_size: usize,
354 nonce_size: usize,
355) -> Result<usize> {
356 plaintext_len
357 .checked_add(tag_size)
358 .and_then(|size| size.checked_add(nonce_size))
359 .ok_or_else(|| {
360 LimboError::Corrupt(format!(
361 "encrypted chunk size overflow: plaintext={plaintext_len}, tag={tag_size}, nonce={nonce_size}"
362 ))
363 })
364}
365
366fn encrypted_payload_blob_size(
369 payload_size: usize,
370 chunk_size: usize,
371 tag_size: usize,
372 nonce_size: usize,
373) -> Result<usize> {
374 let chunk_count = encrypted_payload_chunk_count(payload_size, chunk_size);
375 if chunk_count == 0 {
376 return Ok(0);
377 }
378
379 let full_chunk_on_disk = encrypted_chunk_blob_size(chunk_size, tag_size, nonce_size)?;
380 let full_chunks_total = full_chunk_on_disk
381 .checked_mul(chunk_count.saturating_sub(1))
382 .ok_or_else(|| LimboError::Corrupt("encrypted payload total size overflow".to_string()))?;
383 let last_plaintext_len =
384 encrypted_chunk_plaintext_len(payload_size, chunk_count - 1, chunk_size)?;
385 let last_chunk_on_disk = encrypted_chunk_blob_size(last_plaintext_len, tag_size, nonce_size)?;
386 full_chunks_total
387 .checked_add(last_chunk_on_disk)
388 .ok_or_else(|| LimboError::Corrupt("encrypted payload total size overflow".to_string()))
389}
390
391fn build_encrypted_chunk_aad(
392 salt: u64,
393 payload_size_in_aad: Option<u64>,
394 op_count: u32,
395 commit_ts: u64,
396 chunk_index: u32,
397) -> [u8; ENCRYPTED_CHUNK_AAD_SIZE] {
398 let mut aad = [0u8; ENCRYPTED_CHUNK_AAD_SIZE];
399 aad[..8].copy_from_slice(&salt.to_le_bytes());
400 if let Some(payload_size) = payload_size_in_aad {
401 aad[8..16].copy_from_slice(&payload_size.to_le_bytes());
402 }
403 aad[16..20].copy_from_slice(&op_count.to_le_bytes());
404 aad[20..28].copy_from_slice(&commit_ts.to_le_bytes());
405 aad[28..32].copy_from_slice(&chunk_index.to_le_bytes());
406 aad
407}
408
409#[derive(Clone, Debug)]
411pub struct LogHeader {
412 version: u8,
413 flags: u8,
414 hdr_len: u16,
415 pub(crate) salt: u64,
416 hdr_crc32c: u32,
417 reserved: [u8; LOG_HDR_RESERVED_SIZE],
418}
419
420impl LogHeader {
421 pub(crate) fn new(io: &Arc<dyn crate::IO>) -> Self {
422 Self::new_with_version(io, LOG_VERSION_V2)
423 }
424
425 fn new_with_version(io: &Arc<dyn crate::IO>, version: u8) -> Self {
426 turso_assert!(
427 version == LOG_VERSION_V2 || version == LOG_VERSION,
428 "unsupported logical log header version: {version}"
429 );
430 Self {
431 version,
432 flags: 0,
433 hdr_len: LOG_HDR_SIZE as u16,
434 salt: io.generate_random_number() as u64,
435 hdr_crc32c: 0,
436 reserved: [0; LOG_HDR_RESERVED_SIZE],
437 }
438 }
439
440 fn encode(&self) -> [u8; LOG_HDR_SIZE] {
441 let mut buf = [0u8; LOG_HDR_SIZE];
442 buf[0..4].copy_from_slice(&LOG_MAGIC.to_le_bytes());
443 buf[4] = self.version;
444 buf[5] = self.flags;
445 buf[6..8].copy_from_slice(&self.hdr_len.to_le_bytes());
446 buf[LOG_HDR_SALT_START..LOG_HDR_SALT_START + LOG_HDR_SALT_SIZE]
447 .copy_from_slice(&self.salt.to_le_bytes());
448 buf[LOG_HDR_RESERVED_START..LOG_HDR_CRC_START].copy_from_slice(&self.reserved);
449
450 let crc = crc32c::crc32c(&buf);
451 buf[LOG_HDR_CRC_START..LOG_HDR_SIZE].copy_from_slice(&crc.to_le_bytes());
452 buf
453 }
454
455 fn decode(buf: &[u8]) -> Result<Self> {
456 if buf.len() < LOG_HDR_SIZE {
457 return Err(LimboError::Corrupt(
458 "Logical log header too small".to_string(),
459 ));
460 }
461 let magic = u32::from_le_bytes([buf[0], buf[1], buf[2], buf[3]]);
462 if magic != LOG_MAGIC {
463 return Err(LimboError::Corrupt("Invalid logical log magic".to_string()));
464 }
465 let version = buf[4];
466 if version != LOG_VERSION && version != LOG_VERSION_V2 {
467 return Err(LimboError::Corrupt(format!(
468 "Unsupported logical log version {version}"
469 )));
470 }
471 let flags = buf[5];
472 if flags & 0b1111_1110 != 0 {
473 return Err(LimboError::Corrupt(
474 "Invalid logical log header flags".to_string(),
475 ));
476 }
477 let hdr_len = u16::from_le_bytes([buf[6], buf[7]]);
478 if hdr_len as usize != LOG_HDR_SIZE {
479 return Err(LimboError::Corrupt(format!(
480 "Invalid logical log header length {hdr_len}"
481 )));
482 }
483 if buf.len() < hdr_len as usize {
484 return Err(LimboError::Corrupt(
485 "Logical log header shorter than hdr_len".to_string(),
486 ));
487 }
488 let hdr_crc32c = u32::from_le_bytes([
489 buf[LOG_HDR_CRC_START],
490 buf[LOG_HDR_CRC_START + 1],
491 buf[LOG_HDR_CRC_START + 2],
492 buf[LOG_HDR_CRC_START + 3],
493 ]);
494 let mut crc_buf = [0u8; LOG_HDR_SIZE];
495 crc_buf.copy_from_slice(&buf[..LOG_HDR_SIZE]);
496 crc_buf[LOG_HDR_CRC_START..LOG_HDR_SIZE].fill(0);
497 let expected_crc = crc32c::crc32c(&crc_buf);
498 if expected_crc != hdr_crc32c {
499 return Err(LimboError::Corrupt(
500 "Logical log header checksum mismatch".to_string(),
501 ));
502 }
503
504 let salt = u64::from_le_bytes([
505 buf[LOG_HDR_SALT_START],
506 buf[LOG_HDR_SALT_START + 1],
507 buf[LOG_HDR_SALT_START + 2],
508 buf[LOG_HDR_SALT_START + 3],
509 buf[LOG_HDR_SALT_START + 4],
510 buf[LOG_HDR_SALT_START + 5],
511 buf[LOG_HDR_SALT_START + 6],
512 buf[LOG_HDR_SALT_START + 7],
513 ]);
514
515 let mut reserved = [0u8; LOG_HDR_RESERVED_SIZE];
516 reserved.copy_from_slice(&buf[LOG_HDR_RESERVED_START..LOG_HDR_CRC_START]);
517 if reserved.iter().any(|b| *b != 0) {
518 return Err(LimboError::Corrupt(
519 "Logical log header reserved bytes must be zero".to_string(),
520 ));
521 }
522
523 Ok(Self {
524 version,
525 flags,
526 hdr_len,
527 salt,
528 hdr_crc32c,
529 reserved,
530 })
531 }
532}
533
534fn derive_initial_crc(salt: u64) -> u32 {
537 crc32c::crc32c(&salt.to_le_bytes())
538}
539
540pub struct LogicalLog {
541 pub file: Arc<dyn File>,
542 io: Arc<dyn crate::IO>,
543 pub offset: u64,
544 header: Option<LogHeader>,
545 pub running_crc: u32,
549 pending_running_crc: Option<u32>,
553 encryption_ctx: Option<EncryptionContext>,
554 encrypted_payload_chunk_size: usize,
557 max_appended_commit_ts: u64,
558}
559
560impl LogicalLog {
561 fn new_internal(
562 file: Arc<dyn File>,
563 io: Arc<dyn crate::IO>,
564 encryption_ctx: Option<EncryptionContext>,
565 encrypted_payload_chunk_size: usize,
566 ) -> Self {
567 Self {
568 file,
569 io,
570 offset: 0,
571 header: None,
572 running_crc: 0,
573 pending_running_crc: None,
574 encryption_ctx,
575 encrypted_payload_chunk_size,
576 max_appended_commit_ts: 0,
577 }
578 }
579
580 pub fn new(
581 file: Arc<dyn File>,
582 io: Arc<dyn crate::IO>,
583 encryption_ctx: Option<EncryptionContext>,
584 ) -> Self {
585 Self::new_internal(file, io, encryption_ctx, ENCRYPTED_PAYLOAD_CHUNK_SIZE)
586 }
587
588 #[cfg(clt_turso_tests)]
589 fn new_with_payload_chunk_size(
590 file: Arc<dyn File>,
591 io: Arc<dyn crate::IO>,
592 encryption_ctx: Option<EncryptionContext>,
593 encrypted_payload_chunk_size: usize,
594 ) -> Self {
595 Self::new_internal(file, io, encryption_ctx, encrypted_payload_chunk_size)
596 }
597
598 pub(crate) fn set_header(&mut self, header: LogHeader) {
599 self.running_crc = derive_initial_crc(header.salt);
600 self.header = Some(header);
601 }
602
603 pub(crate) fn header(&self) -> Option<&LogHeader> {
604 self.header.as_ref()
605 }
606
607 pub(crate) fn encryption_ctx(&self) -> Option<&EncryptionContext> {
608 self.encryption_ctx.as_ref()
609 }
610
611 fn frame_and_pwrite_tx(
619 &mut self,
620 mut tx: LogRecord,
621 advance_offset_immediately: bool,
622 on_serialization_complete: OnSerializationComplete<'_>,
623 ) -> Result<(Completion, u64)> {
624 let op_count = tx.op_count;
625 let commit_ts = tx.tx_timestamp;
626 self.max_appended_commit_ts = self.max_appended_commit_ts.max(commit_ts);
627 debug_assert!(
630 tx.buf.len() >= LOG_RECORD_PREFIX_SIZE,
631 "LogRecord buf missing pre-reserved framing prefix"
632 );
633 let payload_size = tx.buf.len() - LOG_RECORD_PREFIX_SIZE;
634 let payload_size_u64 = payload_size as u64;
635
636 #[cfg(clt_turso_feature = "conn_raw_api")]
637 let has_portable_changes = tx.portable_changes_required || !tx.portable_changes.is_empty();
638 #[cfg(not(clt_turso_feature = "conn_raw_api"))]
639 let has_portable_changes = false;
640 #[cfg(clt_turso_feature = "conn_raw_api")]
641 let portable_changes_enabled = tx.portable_changes_enabled || has_portable_changes;
642 #[cfg(not(clt_turso_feature = "conn_raw_api"))]
643 let portable_changes_enabled = false;
644
645 let is_first_write = self.offset == 0;
649 if is_first_write && self.header.is_none() {
650 let version = if portable_changes_enabled {
651 LOG_VERSION
652 } else {
653 LOG_VERSION_V2
654 };
655 let header = LogHeader::new_with_version(&self.io, version);
656 self.running_crc = derive_initial_crc(header.salt);
657 self.header = Some(header);
658 }
659 if portable_changes_enabled {
660 let header = self
661 .header
662 .as_mut()
663 .expect("log header must be set before writing");
664 if header.version == LOG_VERSION_V2 {
665 if !is_first_write {
666 return Err(LimboError::InternalError(
667 "portable logical changes require logical log header upgrade before append"
668 .to_string(),
669 ));
670 }
671 header.version = LOG_VERSION;
672 }
673 }
674 if has_portable_changes {
675 tx.buf.splice(
676 LOG_RECORD_PREFIX_SIZE..LOG_RECORD_PREFIX_SIZE,
677 [0u8; TX_EXT_HEADER_SIZE - TX_HEADER_SIZE],
678 );
679 }
680
681 let tx_header_size = if has_portable_changes {
682 TX_EXT_HEADER_SIZE
683 } else {
684 TX_HEADER_SIZE
685 };
686 let frame_payload_start = LOG_HDR_SIZE + tx_header_size;
687
688 #[cfg(clt_turso_feature = "conn_raw_api")]
689 let extension_block = if !has_portable_changes {
690 Vec::new()
691 } else {
692 let encryption_overhead = self
693 .encryption_ctx
694 .as_ref()
695 .map(|enc_ctx| (enc_ctx.tag_size(), enc_ctx.nonce_size()));
696 let portable_changes = encode_portable_change_payload_with_stable_end_offset(
697 PortableEndOffsetCtx {
698 write_offset: self.offset,
699 includes_log_header: is_first_write,
700 tx_header_size,
701 recovery_payload_size: payload_size,
702 encrypted_payload_chunk_size: self.encrypted_payload_chunk_size,
703 encryption_overhead,
704 },
705 tx.tx_timestamp,
706 &tx.portable_changes,
707 )?;
708 encode_extension_record(EXTENSION_TYPE_PORTABLE_CHANGES, 0, &portable_changes)?
709 };
710 #[cfg(not(clt_turso_feature = "conn_raw_api"))]
711 let extension_block = Vec::new();
712
713 let extension_size = u64::try_from(extension_block.len()).map_err(|_| {
714 LimboError::InternalError("Logical log extension size exceeds u64".to_string())
715 })?;
716 if !extension_block.is_empty() {
717 tx.buf
718 .splice(frame_payload_start..frame_payload_start, extension_block);
719 }
720 let plaintext_size = tx.buf.len() - frame_payload_start;
721 let plaintext_size_u64 = plaintext_size as u64;
722
723 if let Some(enc_ctx) = &self.encryption_ctx {
729 let salt = self
730 .header
731 .as_ref()
732 .expect("log header must be set before writing")
733 .salt;
734 let on_disk_payload_size = encrypted_payload_blob_size(
735 plaintext_size,
736 self.encrypted_payload_chunk_size,
737 enc_ctx.tag_size(),
738 enc_ctx.nonce_size(),
739 )?;
740 let total = frame_payload_start + on_disk_payload_size + TX_TRAILER_SIZE;
741 let plaintext = tx.buf.split_off(frame_payload_start);
745 debug_assert_eq!(plaintext.len(), plaintext_size);
746 tx.buf.reserve(total - tx.buf.len());
747
748 let chunk_count =
749 encrypted_payload_chunk_count(plaintext_size, self.encrypted_payload_chunk_size);
750 let payload_start = tx.buf.len();
751 for (chunk_index, plaintext_chunk) in plaintext
752 .chunks(self.encrypted_payload_chunk_size)
753 .enumerate()
754 {
755 let is_last_chunk = chunk_index + 1 == chunk_count;
756 let aad = build_encrypted_chunk_aad(
757 salt,
758 is_last_chunk.then_some(plaintext_size_u64),
759 op_count,
760 commit_ts,
761 u32::try_from(chunk_index).map_err(|_| {
762 LimboError::InternalError(
763 "encrypted payload chunk index exceeds u32".to_string(),
764 )
765 })?,
766 );
767 let (ciphertext, nonce) = enc_ctx.encrypt_chunk(plaintext_chunk, &aad)?;
768 debug_assert_eq!(
772 ciphertext.len(),
773 plaintext_chunk.len() + enc_ctx.tag_size(),
774 "encrypt_chunk output size mismatch: expected plaintext({}) + tag({}), got {}",
775 plaintext_chunk.len(),
776 enc_ctx.tag_size(),
777 ciphertext.len(),
778 );
779 tx.buf.extend_from_slice(&ciphertext);
780 tx.buf.extend_from_slice(&nonce);
781 }
782 turso_assert!(
783 tx.buf.len() - payload_start == on_disk_payload_size,
784 "encrypted on-disk payload size mismatch"
785 );
786 }
788 let tx_header_start = LOG_HDR_SIZE;
795 let frame_magic = if has_portable_changes {
796 EXT_FRAME_MAGIC
797 } else {
798 FRAME_MAGIC
799 };
800 tx.buf[tx_header_start..tx_header_start + 4].copy_from_slice(&frame_magic.to_le_bytes());
801 tx.buf[tx_header_start + 4..tx_header_start + 12]
802 .copy_from_slice(&payload_size_u64.to_le_bytes());
803 tx.buf[tx_header_start + 12..tx_header_start + 16].copy_from_slice(&op_count.to_le_bytes());
804 tx.buf[tx_header_start + 16..tx_header_start + 24]
805 .copy_from_slice(&commit_ts.to_le_bytes());
806 if has_portable_changes {
807 tx.buf[tx_header_start + 24..tx_header_start + 32]
808 .copy_from_slice(&extension_size.to_le_bytes());
809 tx.buf[tx_header_start + 32..tx_header_start + 36].copy_from_slice(&1u32.to_le_bytes());
810 tx.buf[tx_header_start + 36..tx_header_start + 40]
811 .copy_from_slice(&TX_FRAME_FLAG_HAS_EXTENSION_BLOCK.to_le_bytes());
812 }
813
814 let payload_end = tx.buf.len();
820 let crc = crc32c::crc32c_append(self.running_crc, &tx.buf[tx_header_start..payload_end]);
821 tx.buf.extend_from_slice(&crc.to_le_bytes());
822 tx.buf.extend_from_slice(&END_MAGIC.to_le_bytes());
823
824 if is_first_write {
828 let header_bytes = self.header.as_ref().unwrap().encode();
829 tx.buf[..LOG_HDR_SIZE].copy_from_slice(&header_bytes);
830 }
831
832 let raw = Arc::new(tx.buf.into_boxed_slice());
835 let shared = if is_first_write {
836 SharedBufferData::new(raw)
837 } else {
838 SharedBufferData::new_view(raw, LOG_HDR_SIZE)
839 };
840 if let Some(cb) = on_serialization_complete {
841 cb(shared.clone(), crc)?;
842 }
843
844 let buffer = Arc::new(Buffer::new_shared_data(shared));
849 let buffer_len = buffer.len();
850 let c = Completion::new_write(move |res: Result<i32, CompletionError>| {
851 let Ok(bytes_written) = res else {
852 return;
853 };
854 turso_assert!(
855 bytes_written == buffer_len as i32,
856 "wrote({bytes_written}) != expected({buffer_len})"
857 );
858 });
859
860 let c = self.file.pwrite(self.offset, buffer, c)?;
861 if advance_offset_immediately {
862 self.offset += buffer_len as u64;
863 self.running_crc = crc;
864 } else {
865 self.pending_running_crc = Some(crc);
866 }
867 Ok((c, buffer_len as u64))
868 }
869
870 pub fn log_tx(&mut self, tx: LogRecord) -> Result<Completion> {
873 let (c, _) = self.frame_and_pwrite_tx(tx, true, None)?;
874 Ok(c)
875 }
876
877 pub fn upgrade_header_for_log_tx(&mut self, tx: &LogRecord) -> Result<Option<Completion>> {
878 #[cfg(clt_turso_feature = "conn_raw_api")]
879 let portable_changes_enabled =
880 tx.portable_changes_enabled || !tx.portable_changes.is_empty();
881 #[cfg(not(clt_turso_feature = "conn_raw_api"))]
882 let portable_changes_enabled = {
883 let _ = tx;
884 false
885 };
886
887 if !portable_changes_enabled || self.offset == 0 {
888 return Ok(None);
889 }
890
891 let upgraded_header = {
892 let header = self.header.as_mut().ok_or_else(|| {
893 LimboError::InternalError(
894 "Logical log header not initialized before portable upgrade".to_string(),
895 )
896 })?;
897 if header.version != LOG_VERSION_V2 {
898 return Ok(None);
899 }
900 header.version = LOG_VERSION;
901 header.clone()
902 };
903
904 Ok(Some(self.write_header(upgraded_header)?))
905 }
906
907 pub fn log_tx_deferred_offset(
915 &mut self,
916 tx: LogRecord,
917 on_serialization_complete: OnSerializationComplete<'_>,
918 ) -> Result<(Completion, u64)> {
919 self.frame_and_pwrite_tx(tx, false, on_serialization_complete)
920 }
921
922 pub fn advance_offset_after_success(&mut self, bytes: u64) {
923 self.offset = self
924 .offset
925 .checked_add(bytes)
926 .expect("logical log offset overflow");
927 self.running_crc = self
928 .pending_running_crc
929 .take()
930 .expect("advance_offset_after_success called without pending deferred write");
931 }
932
933 pub fn sync(&mut self, sync_type: FileSyncType) -> Result<Completion> {
934 let completion = Completion::new_sync(move |_| {
935 tracing::debug!("logical_log_sync finish");
936 });
937 let c = self.file.sync(completion, sync_type)?;
938 Ok(c)
939 }
940
941 fn current_or_new_header(&self) -> Result<LogHeader> {
942 if let Some(header) = self.header.clone() {
943 return Ok(header);
944 }
945 if self.offset == 0 {
946 return Ok(LogHeader::new(&self.io));
948 }
949 Err(LimboError::InternalError(
950 "Logical log header not initialized".to_string(),
951 ))
952 }
953
954 fn write_header(&mut self, mut header: LogHeader) -> Result<Completion> {
955 let header_bytes = header.encode();
956 header.hdr_crc32c = u32::from_le_bytes([
957 header_bytes[LOG_HDR_CRC_START],
958 header_bytes[LOG_HDR_CRC_START + 1],
959 header_bytes[LOG_HDR_CRC_START + 2],
960 header_bytes[LOG_HDR_CRC_START + 3],
961 ]);
962 self.header = Some(header);
963
964 let buffer = Arc::new(Buffer::new(header_bytes.to_vec()));
965 let c = Completion::new_write({
966 let buffer_len = buffer.len();
967 move |res: Result<i32, CompletionError>| {
968 let Ok(bytes_written) = res else {
969 return;
970 };
971 turso_assert!(
972 bytes_written == buffer_len as i32,
973 "wrote({bytes_written}) != expected({buffer_len})"
974 );
975 }
976 });
977 self.file.pwrite(0, buffer, c)
978 }
979
980 pub fn update_header(&mut self) -> Result<Completion> {
981 let header = self.current_or_new_header()?;
982 self.write_header(header)
983 }
984
985 fn truncate_to_zero(&mut self) -> Result<Completion> {
986 let mut header = self.current_or_new_header()?;
989 header.salt = self.io.generate_random_number() as u64;
990 self.running_crc = derive_initial_crc(header.salt);
991 self.pending_running_crc = None;
992 self.header = Some(header);
993
994 let completion = Completion::new_trunc(move |result| {
995 if let Err(err) = result {
996 tracing::error!("logical_log_truncate failed: {}", err);
997 }
998 });
999 let c = self.file.truncate(0, completion)?;
1000 self.offset = 0;
1001 self.max_appended_commit_ts = 0;
1002 Ok(c)
1003 }
1004
1005 pub fn truncate(&mut self, checkpointed_through_ts: u64) -> Result<Completion> {
1008 if self.max_appended_commit_ts > checkpointed_through_ts {
1009 let c = Completion::new_trunc(|_| {});
1011 c.complete(0);
1012 return Ok(c);
1013 }
1014 self.truncate_to_zero()
1015 }
1016
1017 pub fn reset_to_fresh_header(&mut self) -> Result<Completion> {
1025 let mut header = self.current_or_new_header()?;
1028 header.salt = self.io.generate_random_number() as u64;
1029 self.running_crc = derive_initial_crc(header.salt);
1030 self.pending_running_crc = None;
1031 self.header = Some(header.clone());
1032
1033 let header_c = self.write_header(header)?;
1034 let truncate_c = self.file.truncate(
1035 LOG_HDR_SIZE as u64,
1036 Completion::new_trunc(move |result| {
1037 if let Err(err) = result {
1038 tracing::error!("logical_log_truncate failed: {}", err);
1039 }
1040 }),
1041 )?;
1042 self.offset = 0;
1043
1044 let mut group = CompletionGroup::new(|_| {});
1045 group.add(&header_c);
1046 group.add(&truncate_c);
1047 Ok(group.build())
1048 }
1049}
1050
1051pub(crate) fn serialize_op_entry(
1054 buffer: &mut Vec<u8>,
1055 row_version: &RowVersion,
1056 portable_extension: Option<&[u8]>,
1057) -> Result<()> {
1058 let is_delete = row_version.end().is_some();
1059
1060 let mut flags = 0u8;
1061 if row_version.btree_resident {
1062 flags |= OP_FLAG_BTREE_RESIDENT;
1063 }
1064 if portable_extension.is_some_and(|extension| !extension.is_empty()) {
1065 flags |= OP_FLAG_PORTABLE_EXTENSION;
1066 }
1067
1068 let table_id_i64: i64 = row_version.row.id.table_id.into();
1069 turso_assert!(
1070 table_id_i64 < 0,
1071 "table_id_i64 should be negative, but got {table_id_i64}"
1072 );
1073 turso_assert!(
1074 (i32::MIN as i64..=i32::MAX as i64).contains(&table_id_i64),
1075 "table_id_i64 out of i32 range: {table_id_i64}"
1076 );
1077 let table_id_i32 = table_id_i64 as i32;
1078
1079 let write_header = |buf: &mut Vec<u8>, tag: u8| {
1080 buf.push(tag);
1081 buf.push(flags);
1082 buf.extend_from_slice(&table_id_i32.to_le_bytes());
1083 };
1084
1085 match (&row_version.row.id.row_id, is_delete) {
1086 (&RowKey::Int(rowid), false) => {
1087 write_header(buffer, OP_UPSERT_TABLE);
1088 let record_bytes = row_version.row.payload();
1089 let rowid_u64 = rowid as u64;
1090 let rowid_len = varint_len(rowid_u64);
1091 let payload_len = rowid_len + record_bytes.len();
1092 write_varint_to_vec(payload_len as u64, buffer);
1093 write_varint_to_vec(rowid_u64, buffer);
1094 buffer.extend_from_slice(record_bytes);
1095 }
1096 (&RowKey::Int(rowid), true) => {
1097 write_header(buffer, OP_DELETE_TABLE);
1098 let rowid_u64 = rowid as u64;
1099 let rowid_len = varint_len(rowid_u64);
1100 write_varint_to_vec(rowid_len as u64, buffer);
1101 write_varint_to_vec(rowid_u64, buffer);
1102 }
1103 (RowKey::Record(_), is_delete) => {
1104 write_header(
1105 buffer,
1106 if is_delete {
1107 OP_DELETE_INDEX
1108 } else {
1109 OP_UPSERT_INDEX
1110 },
1111 );
1112 let key_bytes = row_version.row.payload();
1113 write_varint_to_vec(key_bytes.len() as u64, buffer);
1114 buffer.extend_from_slice(key_bytes);
1115 }
1116 }
1117
1118 if let Some(portable_extension) =
1119 portable_extension.filter(|portable_extension| !portable_extension.is_empty())
1120 {
1121 write_varint_to_vec(portable_extension.len() as u64, buffer);
1122 buffer.extend_from_slice(portable_extension);
1123 }
1124
1125 Ok(())
1126}
1127
1128pub(crate) fn serialize_header_entry(buffer: &mut Vec<u8>, header: &DatabaseHeader) {
1129 buffer.push(OP_UPDATE_HEADER);
1131 buffer.push(0);
1132 buffer.extend_from_slice(&0i32.to_le_bytes());
1133 write_varint_to_vec(DatabaseHeader::SIZE as u64, buffer);
1134 buffer.extend_from_slice(bytemuck::bytes_of(header));
1135}
1136
1137fn write_proto_varint(mut value: u64, buffer: &mut Vec<u8>) {
1138 while value >= 0x80 {
1139 buffer.push((value as u8) | 0x80);
1140 value >>= 7;
1141 }
1142 buffer.push(value as u8);
1143}
1144
1145fn write_proto_key(field: u64, wire_type: u64, buffer: &mut Vec<u8>) {
1146 write_proto_varint((field << 3) | wire_type, buffer);
1147}
1148
1149fn write_proto_sint64(field: u64, value: i64, buffer: &mut Vec<u8>) {
1150 let zigzag = ((value << 1) ^ (value >> 63)) as u64;
1151 write_proto_key(field, 0, buffer);
1152 write_proto_varint(zigzag, buffer);
1153}
1154
1155fn write_proto_bytes(field: u64, value: &[u8], buffer: &mut Vec<u8>) {
1156 write_proto_key(field, 2, buffer);
1157 write_proto_varint(value.len() as u64, buffer);
1158 buffer.extend_from_slice(value);
1159}
1160
1161pub(crate) fn encode_delete_portable_extension(
1162 identity_record: Option<&[u8]>,
1163 pk_record: Option<&[u8]>,
1164 rowid: Option<i64>,
1165) -> Vec<u8> {
1166 let mut extension = Vec::new();
1167 if let Some(identity_record) = identity_record.filter(|record| !record.is_empty()) {
1168 write_proto_bytes(
1169 OP_EXT_FIELD_DELETE_IDENTITY_RECORD,
1170 identity_record,
1171 &mut extension,
1172 );
1173 }
1174 if let Some(pk_record) = pk_record.filter(|record| !record.is_empty()) {
1175 write_proto_bytes(OP_EXT_FIELD_DELETE_PK_RECORD, pk_record, &mut extension);
1176 }
1177 if let Some(rowid) = rowid {
1178 write_proto_sint64(OP_EXT_FIELD_DELETE_ROWID, rowid, &mut extension);
1179 }
1180 extension
1181}
1182
1183fn read_proto_varint_from_buf(bytes: &[u8], offset: &mut usize) -> Result<u64> {
1184 let mut value = 0u64;
1185 let mut shift = 0;
1186 while *offset < bytes.len() {
1187 let byte = bytes[*offset];
1188 *offset += 1;
1189 value |= ((byte & 0x7f) as u64) << shift;
1190 if byte & 0x80 == 0 {
1191 return Ok(value);
1192 }
1193 shift += 7;
1194 if shift >= 64 {
1195 return Err(LimboError::Corrupt("protobuf varint overflows u64".into()));
1196 }
1197 }
1198 Err(LimboError::Corrupt("truncated protobuf varint".into()))
1199}
1200
1201fn skip_proto_field(bytes: &[u8], offset: &mut usize, wire_type: u64) -> Result<()> {
1202 match wire_type {
1203 0 => {
1204 let _ = read_proto_varint_from_buf(bytes, offset)?;
1205 }
1206 2 => {
1207 let len = read_proto_varint_from_buf(bytes, offset)?;
1208 let len = usize::try_from(len)
1209 .map_err(|_| LimboError::Corrupt("protobuf field length overflows usize".into()))?;
1210 let end = offset
1211 .checked_add(len)
1212 .ok_or_else(|| LimboError::Corrupt("protobuf field length overflow".into()))?;
1213 if end > bytes.len() {
1214 return Err(LimboError::Corrupt(
1215 "protobuf length-delimited field exceeds extension".into(),
1216 ));
1217 }
1218 *offset = end;
1219 }
1220 other => {
1221 return Err(LimboError::Corrupt(format!(
1222 "unsupported protobuf wire type in op extension: {other}"
1223 )));
1224 }
1225 }
1226 Ok(())
1227}
1228
1229fn read_proto_sint64_from_buf(bytes: &[u8], offset: &mut usize) -> Result<i64> {
1230 let value = read_proto_varint_from_buf(bytes, offset)?;
1231 Ok(((value >> 1) as i64) ^ (-((value & 1) as i64)))
1232}
1233
1234fn decode_delete_portable_extension(extension: &[u8]) -> Result<DeletePortableExtension> {
1235 let mut offset = 0usize;
1236 let mut decoded = DeletePortableExtension::default();
1237 while offset < extension.len() {
1238 let key = read_proto_varint_from_buf(extension, &mut offset)?;
1239 let field = key >> 3;
1240 let wire_type = key & 7;
1241 match (field, wire_type) {
1242 (OP_EXT_FIELD_DELETE_IDENTITY_RECORD, 2) => {
1243 let len = read_proto_varint_from_buf(extension, &mut offset)?;
1244 let len = usize::try_from(len).map_err(|_| {
1245 LimboError::Corrupt("delete identity record length overflows usize".into())
1246 })?;
1247 let end = offset.checked_add(len).ok_or_else(|| {
1248 LimboError::Corrupt("delete identity record length overflow".into())
1249 })?;
1250 if end > extension.len() {
1251 return Err(LimboError::Corrupt(
1252 "delete identity record exceeds op extension".into(),
1253 ));
1254 }
1255 decoded.identity_record = extension[offset..end].to_vec();
1256 offset = end;
1257 }
1258 (OP_EXT_FIELD_DELETE_PK_RECORD, 2) => {
1259 let len = read_proto_varint_from_buf(extension, &mut offset)?;
1260 let len = usize::try_from(len).map_err(|_| {
1261 LimboError::Corrupt("delete PK record length overflows usize".into())
1262 })?;
1263 let end = offset.checked_add(len).ok_or_else(|| {
1264 LimboError::Corrupt("delete PK record length overflow".into())
1265 })?;
1266 if end > extension.len() {
1267 return Err(LimboError::Corrupt(
1268 "delete PK record exceeds op extension".into(),
1269 ));
1270 }
1271 decoded.pk_record = extension[offset..end].to_vec();
1272 offset = end;
1273 }
1274 (OP_EXT_FIELD_DELETE_ROWID, 0) => {
1275 let _ = read_proto_sint64_from_buf(extension, &mut offset)?;
1276 }
1277 _ => skip_proto_field(extension, &mut offset, wire_type)?,
1278 }
1279 }
1280 Ok(decoded)
1281}
1282
1283fn proto_varint_len(mut value: u64) -> usize {
1284 let mut len = 1;
1285 while value >= 0x80 {
1286 len += 1;
1287 value >>= 7;
1288 }
1289 len
1290}
1291
1292fn encode_portable_change_payload(
1293 end_offset: u64,
1294 commit_ts: u64,
1295 encoded_metadata: &[u8],
1296) -> Vec<u8> {
1297 let body_len =
1298 2 + proto_varint_len(end_offset) + proto_varint_len(commit_ts) + encoded_metadata.len();
1299 let mut out = Vec::with_capacity(proto_varint_len(body_len as u64) + body_len);
1300 write_proto_varint(body_len as u64, &mut out);
1301 write_proto_varint(1 << 3, &mut out);
1303 write_proto_varint(end_offset, &mut out);
1304 write_proto_varint(2 << 3, &mut out);
1306 write_proto_varint(commit_ts, &mut out);
1307 out.extend_from_slice(encoded_metadata);
1308 out
1309}
1310
1311struct PortableEndOffsetCtx {
1319 write_offset: u64,
1320 includes_log_header: bool,
1321 tx_header_size: usize,
1322 recovery_payload_size: usize,
1323 encrypted_payload_chunk_size: usize,
1324 encryption_overhead: Option<(usize, usize)>,
1325}
1326
1327fn encode_portable_change_payload_with_stable_end_offset(
1328 ctx: PortableEndOffsetCtx,
1329 tx_timestamp: u64,
1330 portable_changes: &[u8],
1331) -> Result<Vec<u8>> {
1332 let frame_end_offset = |portable_payload_len: usize| -> Result<u64> {
1333 let extension_size = EXTENSION_RECORD_HEADER_SIZE
1334 .checked_add(portable_payload_len)
1335 .ok_or_else(|| {
1336 LimboError::InternalError("portable logical extension size overflow".to_string())
1337 })?;
1338 let plaintext_size = ctx
1339 .recovery_payload_size
1340 .checked_add(extension_size)
1341 .ok_or_else(|| {
1342 LimboError::InternalError("portable logical plaintext size overflow".to_string())
1343 })?;
1344 let body_size = if let Some((tag_size, nonce_size)) = ctx.encryption_overhead {
1345 encrypted_payload_blob_size(
1346 plaintext_size,
1347 ctx.encrypted_payload_chunk_size,
1348 tag_size,
1349 nonce_size,
1350 )?
1351 } else {
1352 plaintext_size
1353 };
1354 let prefix_size = if ctx.includes_log_header {
1355 LOG_HDR_SIZE
1356 } else {
1357 0
1358 };
1359 let frame_bytes = prefix_size
1360 .checked_add(ctx.tx_header_size)
1361 .and_then(|value| value.checked_add(body_size))
1362 .and_then(|value| value.checked_add(TX_TRAILER_SIZE))
1363 .ok_or_else(|| {
1364 LimboError::InternalError("portable logical frame size overflow".to_string())
1365 })?;
1366 ctx.write_offset
1367 .checked_add(frame_bytes as u64)
1368 .ok_or_else(|| {
1369 LimboError::InternalError("portable logical frame offset overflow".to_string())
1370 })
1371 };
1372
1373 let mut end_offset = frame_end_offset(0)?;
1374 loop {
1375 let payload = encode_portable_change_payload(end_offset, tx_timestamp, portable_changes);
1376 let next_end_offset = frame_end_offset(payload.len())?;
1377 if next_end_offset == end_offset {
1378 return Ok(payload);
1379 }
1380 end_offset = next_end_offset;
1381 }
1382}
1383
1384fn encode_extension_record(
1385 extension_type: u16,
1386 extension_flags: u16,
1387 payload: &[u8],
1388) -> Result<Vec<u8>> {
1389 let payload_len = u32::try_from(payload.len()).map_err(|_| {
1390 LimboError::InternalError("Logical log extension record exceeds u32".to_string())
1391 })?;
1392 let mut record = Vec::with_capacity(EXTENSION_RECORD_HEADER_SIZE + payload.len());
1393 record.extend_from_slice(&extension_type.to_le_bytes());
1394 record.extend_from_slice(&extension_flags.to_le_bytes());
1395 record.extend_from_slice(&payload_len.to_le_bytes());
1396 record.extend_from_slice(payload);
1397 Ok(record)
1398}
1399
1400fn find_extension_payload(
1401 extension_block: &[u8],
1402 extension_record_count: u32,
1403 wanted_type: u16,
1404) -> Result<Vec<u8>> {
1405 let mut offset = 0usize;
1406 let mut payload = Vec::new();
1407 for _ in 0..extension_record_count {
1408 let Some(header_end) = offset.checked_add(EXTENSION_RECORD_HEADER_SIZE) else {
1409 return Err(LimboError::Corrupt(
1410 "extension record header offset overflow".to_string(),
1411 ));
1412 };
1413 if header_end > extension_block.len() {
1414 return Err(LimboError::Corrupt(
1415 "extension record header is truncated".to_string(),
1416 ));
1417 }
1418 let extension_type =
1419 u16::from_le_bytes(extension_block[offset..offset + 2].try_into().unwrap());
1420 let extension_flags =
1421 u16::from_le_bytes(extension_block[offset + 2..offset + 4].try_into().unwrap());
1422 if extension_flags != 0 {
1423 return Err(LimboError::Corrupt(format!(
1424 "unsupported extension flags for type {extension_type}: {extension_flags:#x}"
1425 )));
1426 }
1427 let extension_len = u32::from_le_bytes(
1428 extension_block[offset + 4..offset + EXTENSION_RECORD_HEADER_SIZE]
1429 .try_into()
1430 .unwrap(),
1431 ) as usize;
1432 let payload_start = header_end;
1433 let Some(payload_end) = payload_start.checked_add(extension_len) else {
1434 return Err(LimboError::Corrupt(
1435 "extension record payload offset overflow".to_string(),
1436 ));
1437 };
1438 if payload_end > extension_block.len() {
1439 return Err(LimboError::Corrupt(
1440 "extension record payload is truncated".to_string(),
1441 ));
1442 }
1443 if extension_type == wanted_type {
1444 payload.extend_from_slice(&extension_block[payload_start..payload_end]);
1445 }
1446 offset = payload_end;
1447 }
1448 if offset != extension_block.len() {
1449 return Err(LimboError::Corrupt(
1450 "extension block has trailing bytes".to_string(),
1451 ));
1452 }
1453 Ok(payload)
1454}
1455
1456pub(crate) fn parse_ops_from_plaintext(
1459 plaintext: &[u8],
1460 payload_size: usize,
1461 op_count: u32,
1462 commit_ts: u64,
1463) -> Result<Vec<ParsedOp>> {
1464 if plaintext.len() != payload_size {
1465 return Err(LimboError::Corrupt(format!(
1466 "decrypted size ({}) != payload_size ({payload_size})",
1467 plaintext.len()
1468 )));
1469 }
1470 let mut ops = Vec::with_capacity((op_count as usize).min(1024));
1471 let mut cursor = 0usize;
1472 for _ in 0..op_count {
1473 match try_parse_one_op_from_buf(&plaintext[cursor..], commit_ts)? {
1474 Some((op, consumed)) => {
1475 cursor += consumed;
1476 ops.push(op);
1477 }
1478 None => {
1479 return Err(LimboError::Corrupt(
1480 "incomplete op in decrypted payload".into(),
1481 ));
1482 }
1483 }
1484 }
1485 if cursor != plaintext.len() {
1486 return Err(LimboError::Corrupt(format!(
1487 "trailing bytes after ops: consumed {cursor}, total {}",
1488 plaintext.len()
1489 )));
1490 }
1491 Ok(ops)
1492}
1493
1494fn try_parse_one_op_from_buf(buf: &[u8], commit_ts: u64) -> Result<Option<(ParsedOp, usize)>> {
1500 if buf.len() < 6 {
1501 return Ok(None);
1502 }
1503
1504 let tag = buf[0];
1505 let flags = buf[1];
1506 let table_id_i32 = i32::from_le_bytes([buf[2], buf[3], buf[4], buf[5]]);
1507
1508 let table_id: Option<MVTableId> = match tag {
1509 OP_UPSERT_TABLE | OP_DELETE_TABLE | OP_UPSERT_INDEX | OP_DELETE_INDEX => {
1510 if flags & !OP_ALLOWED_FLAGS != 0 || table_id_i32 >= 0 {
1511 return Err(LimboError::Corrupt(
1512 "Invalid op flags or non-negative table_id".into(),
1513 ));
1514 }
1515 Some(MVTableId::from(table_id_i32 as i64))
1516 }
1517 OP_UPDATE_HEADER => {
1518 if flags != 0 || table_id_i32 != 0 {
1519 return Err(LimboError::Corrupt(
1520 "Invalid UPDATE_HEADER flags/table_id".into(),
1521 ));
1522 }
1523 None
1524 }
1525 _ => return Err(LimboError::Corrupt(format!("Unknown op tag: {tag}"))),
1526 };
1527 let btree_resident = (flags & OP_FLAG_BTREE_RESIDENT) != 0;
1528
1529 let Some((payload_len_u64, varint_bytes)) = read_varint_partial(&buf[6..])? else {
1530 return Ok(None);
1531 };
1532 let payload_len = match usize::try_from(payload_len_u64) {
1533 Ok(v) => v,
1534 Err(_) => return Err(LimboError::Corrupt("payload_len overflows usize".into())),
1535 };
1536
1537 let fixed = 6 + varint_bytes;
1538 let total = fixed + payload_len;
1539 if buf.len() < total {
1540 return Ok(None);
1541 }
1542
1543 let payload = &buf[fixed..total];
1544 let (extension, total) = if flags & OP_FLAG_PORTABLE_EXTENSION == 0 {
1545 (&[][..], total)
1546 } else {
1547 let Some((extension_len_u64, extension_len_bytes)) = read_varint_partial(&buf[total..])?
1548 else {
1549 return Ok(None);
1550 };
1551 let extension_len = usize::try_from(extension_len_u64)
1552 .map_err(|_| LimboError::Corrupt("op extension length overflows usize".into()))?;
1553 let extension_start = total + extension_len_bytes;
1554 let extension_end = extension_start
1555 .checked_add(extension_len)
1556 .ok_or_else(|| LimboError::Corrupt("op extension length overflow".into()))?;
1557 if buf.len() < extension_end {
1558 return Ok(None);
1559 }
1560 (&buf[extension_start..extension_end], extension_end)
1561 };
1562
1563 let parsed_op = match tag {
1564 OP_UPSERT_TABLE => {
1565 let table_id = table_id.expect("table op must have table_id");
1566 let (rowid_u64, rowid_len) = read_varint(payload)
1567 .map_err(|_| LimboError::Corrupt("Bad rowid varint in UPSERT_TABLE".into()))?;
1568 if rowid_len > payload.len() {
1569 return Err(LimboError::Corrupt("rowid_len > payload".into()));
1570 }
1571 let record_bytes = payload[rowid_len..].to_vec();
1572 let rowid = RowID::new(table_id, RowKey::Int(rowid_u64 as i64));
1573 ParsedOp::UpsertTable {
1574 table_id,
1575 rowid,
1576 record_bytes,
1577 commit_ts,
1578 btree_resident,
1579 }
1580 }
1581 OP_DELETE_TABLE => {
1582 let table_id = table_id.expect("table op must have table_id");
1583 let (rowid_u64, rowid_len) = read_varint(payload)
1584 .map_err(|_| LimboError::Corrupt("Bad rowid varint in DELETE_TABLE".into()))?;
1585 if rowid_len > payload.len() {
1586 return Err(LimboError::Corrupt(
1587 "DELETE_TABLE payload size mismatch".into(),
1588 ));
1589 }
1590 let mut record_bytes = payload[rowid_len..].to_vec();
1591 let mut pk_record_bytes = Vec::new();
1592 if !extension.is_empty() {
1593 let decoded = decode_delete_portable_extension(extension)?;
1594 if record_bytes.is_empty() {
1595 record_bytes = decoded.identity_record;
1596 }
1597 pk_record_bytes = decoded.pk_record;
1598 }
1599 let rowid = RowID::new(table_id, RowKey::Int(rowid_u64 as i64));
1600 ParsedOp::DeleteTable {
1601 rowid,
1602 record_bytes,
1603 pk_record_bytes,
1604 commit_ts,
1605 btree_resident,
1606 }
1607 }
1608 OP_UPSERT_INDEX => ParsedOp::UpsertIndex {
1609 table_id: table_id.expect("index op must have table_id"),
1610 payload: payload.to_vec(),
1611 commit_ts,
1612 btree_resident,
1613 },
1614 OP_DELETE_INDEX => ParsedOp::DeleteIndex {
1615 table_id: table_id.expect("index op must have table_id"),
1616 payload: payload.to_vec(),
1617 commit_ts,
1618 btree_resident,
1619 },
1620 OP_UPDATE_HEADER => {
1621 if payload.len() != DatabaseHeader::SIZE {
1622 return Err(LimboError::Corrupt(
1623 "UPDATE_HEADER wrong payload size".into(),
1624 ));
1625 }
1626 let mut bytes = [0u8; DatabaseHeader::SIZE];
1627 bytes.copy_from_slice(payload);
1628 let header = *bytemuck::from_bytes::<DatabaseHeader>(&bytes);
1629 if header.magic != *b"SQLite format 3\0" {
1630 return Err(LimboError::Corrupt("UPDATE_HEADER bad SQLite magic".into()));
1631 }
1632 ParsedOp::UpdateHeader { header, commit_ts }
1633 }
1634 _ => unreachable!("tag validated above"),
1635 };
1636
1637 Ok(Some((parsed_op, total)))
1638}
1639
1640#[derive(Debug)]
1641pub enum StreamingResult {
1642 UpsertTableRow {
1643 row: Row,
1644 rowid: RowID,
1645 commit_ts: u64,
1646 btree_resident: bool,
1647 },
1648 DeleteTableRow {
1649 rowid: RowID,
1650 commit_ts: u64,
1651 btree_resident: bool,
1652 },
1653 UpsertIndexRow {
1654 row: Row,
1655 rowid: RowID,
1656 commit_ts: u64,
1657 btree_resident: bool,
1658 },
1659 DeleteIndexRow {
1660 row: Row,
1661 rowid: RowID,
1662 commit_ts: u64,
1663 btree_resident: bool,
1664 },
1665 UpdateHeader {
1666 header: DatabaseHeader,
1667 commit_ts: u64,
1668 },
1669 Eof,
1670}
1671
1672#[derive(Debug, Clone, PartialEq, Eq)]
1673pub struct PortableChangeFrame {
1674 pub end_offset: u64,
1675 pub commit_ts: u64,
1676 pub extension_record_count: u32,
1677 pub payload: Vec<u8>,
1678}
1679
1680#[derive(Clone, Copy, Debug)]
1681enum StreamingState {
1682 NeedTransactionStart,
1683}
1684
1685#[derive(Clone, Copy, PartialEq, Eq, Debug)]
1689enum FramePhase {
1690 Header,
1691 ExtensionBlock,
1692 Payload,
1693 Trailer,
1694}
1695
1696struct FrameInProgress {
1714 frame_start: usize,
1715 phase: FramePhase,
1716 payload_size: usize,
1718 op_count: u32,
1719 commit_ts: u64,
1720 extension_size: usize,
1721 extension_record_count: u32,
1722 frame_flags: u32,
1723 running_crc: u32,
1725 parsed_ops: Vec<ParsedOp>,
1726 portable_changes: Vec<u8>,
1727 payload_bytes_read: u64,
1728 op_index: u32,
1729}
1730
1731impl FrameInProgress {
1732 fn new(frame_start: usize) -> Self {
1733 Self {
1734 frame_start,
1735 phase: FramePhase::Header,
1736 payload_size: 0,
1737 op_count: 0,
1738 commit_ts: 0,
1739 extension_size: 0,
1740 extension_record_count: 0,
1741 frame_flags: 0,
1742 running_crc: 0,
1743 parsed_ops: Vec::new(),
1744 portable_changes: Vec::new(),
1745 payload_bytes_read: 0,
1746 op_index: 0,
1747 }
1748 }
1749}
1750
1751struct FrameHeader {
1753 payload_size: usize,
1754 op_count: u32,
1755 commit_ts: u64,
1756 extension_size: usize,
1757 extension_record_count: u32,
1758 frame_flags: u32,
1759 running_crc: u32,
1761}
1762
1763enum HeaderParseOutcome {
1765 Ok(FrameHeader),
1766 Eof,
1767 Invalid,
1768}
1769
1770enum PayloadOutcome {
1774 Ok,
1775 Eof,
1776}
1777
1778#[derive(Debug, Clone)]
1780pub enum HeaderReadResult {
1781 Valid(LogHeader),
1783 NoLog,
1785 Invalid,
1787}
1788
1789#[derive(Debug)]
1794enum InFlightRead {
1795 Chunk {
1799 completion: Completion,
1800 pre_size: usize,
1801 },
1802 Exact {
1804 completion: Completion,
1805 out: Arc<RwLock<Vec<u8>>>,
1806 expected_len: usize,
1807 },
1808}
1809
1810type ReadEncryptedResult = Option<(Vec<u8>, u32)>;
1812
1813pub struct StreamingLogicalLogReader {
1814 file: Arc<dyn File>,
1815 pub offset: usize,
1817 header: Option<LogHeader>,
1819 buffer: Arc<RwLock<Vec<u8>>>,
1821 buffer_offset: usize,
1823 frame_anchor: usize,
1831 file_size: usize,
1832 state: StreamingState,
1833 last_valid_offset: usize,
1836 running_crc: u32,
1839 encryption_ctx: Option<EncryptionContext>,
1840 encrypted_payload_chunk_size: usize,
1843 #[cfg(clt_turso_tests)]
1844 pending_ops: std::collections::VecDeque<ParsedOp>,
1845 decrypt_scratch: Vec<u8>,
1848 in_flight_read: Option<InFlightRead>,
1851 frame_in_progress: Option<FrameInProgress>,
1856}
1857
1858impl StreamingLogicalLogReader {
1859 fn new_internal(
1860 file: Arc<dyn File>,
1861 encryption_ctx: Option<EncryptionContext>,
1862 encrypted_payload_chunk_size: usize,
1863 ) -> Self {
1864 let file_size = file.size().expect("failed to get file size") as usize;
1865 let decrypt_scratch = encryption_ctx
1866 .as_ref()
1867 .map(|enc_ctx| Vec::with_capacity(encrypted_payload_chunk_size + enc_ctx.tag_size()))
1868 .unwrap_or_default();
1869 Self {
1870 file,
1871 offset: 0,
1872 header: None,
1873 buffer: Arc::new(RwLock::new(Vec::with_capacity(4096))),
1874 buffer_offset: 0,
1875 frame_anchor: 0,
1876 file_size,
1877 state: StreamingState::NeedTransactionStart,
1878 last_valid_offset: 0,
1879 running_crc: 0,
1880 encryption_ctx,
1881 encrypted_payload_chunk_size,
1882 #[cfg(clt_turso_tests)]
1883 pending_ops: std::collections::VecDeque::new(),
1884 decrypt_scratch,
1885 in_flight_read: None,
1886 frame_in_progress: None,
1887 }
1888 }
1889
1890 pub fn new(file: Arc<dyn File>, encryption_ctx: Option<EncryptionContext>) -> Self {
1891 Self::new_internal(file, encryption_ctx, ENCRYPTED_PAYLOAD_CHUNK_SIZE)
1892 }
1893
1894 #[cfg(clt_turso_tests)]
1895 fn new_with_payload_chunk_size(
1896 file: Arc<dyn File>,
1897 encryption_ctx: Option<EncryptionContext>,
1898 encrypted_payload_chunk_size: usize,
1899 ) -> Self {
1900 Self::new_internal(file, encryption_ctx, encrypted_payload_chunk_size)
1901 }
1902
1903 pub(crate) fn header(&self) -> Option<&LogHeader> {
1904 self.header.as_ref()
1905 }
1906
1907 pub fn last_valid_offset(&self) -> usize {
1911 self.last_valid_offset
1912 }
1913
1914 #[cfg(clt_turso_tests)]
1915 pub fn has_pending_ops(&self) -> bool {
1916 !self.pending_ops.is_empty()
1917 }
1918
1919 pub fn running_crc(&self) -> u32 {
1922 self.running_crc
1923 }
1924
1925 fn tx_min_frame_size(&self) -> usize {
1926 match self.header.as_ref().map(|header| header.version) {
1927 Some(LOG_VERSION_V2) => TX_MIN_FRAME_SIZE_V2,
1928 _ => TX_MIN_FRAME_SIZE,
1929 }
1930 }
1931
1932 pub fn read_header(&mut self, io: &Arc<dyn crate::IO>) -> Result<()> {
1933 match self.try_read_header(io)? {
1934 HeaderReadResult::Valid(_) => Ok(()),
1935 HeaderReadResult::NoLog => Err(LimboError::Corrupt(
1936 "Logical log header incomplete".to_string(),
1937 )),
1938 HeaderReadResult::Invalid => Err(LimboError::Corrupt(
1939 "Logical log header corrupt".to_string(),
1940 )),
1941 }
1942 }
1943
1944 pub(crate) fn try_read_header(&mut self, io: &Arc<dyn crate::IO>) -> Result<HeaderReadResult> {
1950 let io = io.clone();
1951 io.block(|| self.try_read_header_nonblock())
1952 }
1953
1954 pub(crate) fn try_read_header_nonblock(&mut self) -> Result<IOResult<HeaderReadResult>> {
1955 self.file_size = self.file.size()? as usize;
1956 if self.file_size < LOG_HDR_SIZE {
1957 return Ok(IOResult::Done(HeaderReadResult::NoLog));
1958 }
1959
1960 let header_bytes = return_if_io!(self.read_exact_at(0, LOG_HDR_SIZE));
1961 if header_bytes.iter().all(|&b| b == 0) {
1963 return Ok(IOResult::Done(HeaderReadResult::NoLog));
1964 }
1965 let hdr_len = u16::from_le_bytes([header_bytes[6], header_bytes[7]]) as usize;
1966 if hdr_len != LOG_HDR_SIZE {
1967 self.set_invalid_header_state();
1968 return Ok(IOResult::Done(HeaderReadResult::Invalid));
1969 }
1970
1971 match LogHeader::decode(&header_bytes) {
1972 Ok(header) => {
1973 self.running_crc = derive_initial_crc(header.salt);
1974 self.header = Some(header.clone());
1975 self.offset = hdr_len;
1976 self.buffer.write().clear();
1977 self.buffer_offset = 0;
1978 self.frame_anchor = 0;
1979 self.frame_in_progress = None;
1980 self.last_valid_offset = hdr_len;
1981 Ok(IOResult::Done(HeaderReadResult::Valid(header)))
1982 }
1983 Err(LimboError::Corrupt(_)) => {
1984 self.set_invalid_header_state();
1985 Ok(IOResult::Done(HeaderReadResult::Invalid))
1986 }
1987 Err(err) => Err(err),
1988 }
1989 }
1990
1991 fn set_invalid_header_state(&mut self) {
1992 self.header = None;
1993 self.offset = LOG_HDR_SIZE;
1994 self.buffer.write().clear();
1995 self.buffer_offset = 0;
1996 self.frame_anchor = 0;
1997 self.frame_in_progress = None;
1998 self.last_valid_offset = LOG_HDR_SIZE;
1999 }
2000
2001 #[cfg(clt_turso_tests)]
2002 pub(crate) fn next_frame_blocking(
2003 &mut self,
2004 io: &Arc<dyn crate::IO>,
2005 ) -> Result<Option<Vec<ParsedOp>>> {
2006 let io = io.clone();
2007 io.block(|| self.next_frame())
2008 }
2009
2010 pub(crate) fn next_frame(&mut self) -> Result<IOResult<Option<Vec<ParsedOp>>>> {
2016 loop {
2017 match self.state {
2018 StreamingState::NeedTransactionStart => {
2019 if self.frame_in_progress.is_none() {
2028 self.buffer_offset = self.frame_anchor;
2029 if self.remaining_bytes() < TX_MIN_FRAME_SIZE {
2030 return Ok(IOResult::Done(None));
2031 }
2032 }
2033
2034 let ops = match return_if_io!(self.parse_next_transaction()) {
2035 ParseResult::Frame(frame) => frame.ops,
2036 ParseResult::Eof | ParseResult::InvalidFrame => {
2037 return Ok(IOResult::Done(None))
2038 }
2039 };
2040
2041 if ops.is_empty() {
2042 continue;
2043 }
2044 return Ok(IOResult::Done(Some(ops)));
2045 }
2046 }
2047 }
2048 }
2049
2050 #[cfg(clt_turso_tests)]
2055 pub fn next_record(
2056 &mut self,
2057 io: &Arc<dyn crate::IO>,
2058 mut get_index_info: impl FnMut(MVTableId) -> Result<Arc<IndexInfo>>,
2059 ) -> Result<StreamingResult> {
2060 let mut get_index_info = |index_id, _op_kind| get_index_info(index_id);
2061 self.file_size = self.file.size()? as usize;
2062 if let Some(op) = self.pending_ops.pop_front() {
2063 return self.parsed_op_to_streaming(op, &mut get_index_info);
2064 }
2065
2066 loop {
2067 match self.state {
2068 StreamingState::NeedTransactionStart => {
2069 if self.remaining_bytes() < self.tx_min_frame_size() {
2070 return Ok(StreamingResult::Eof);
2071 }
2072
2073 let ops = match io.block(|| self.parse_next_transaction())? {
2074 ParseResult::Frame(frame) => frame.ops,
2075 ParseResult::Eof | ParseResult::InvalidFrame => {
2076 return Ok(StreamingResult::Eof);
2077 }
2078 };
2079
2080 if ops.is_empty() {
2081 continue;
2082 }
2083 self.pending_ops = ops.into();
2084 let op = self
2085 .pending_ops
2086 .pop_front()
2087 .expect("ops queue should not be empty");
2088 return self.parsed_op_to_streaming(op, &mut get_index_info);
2089 }
2090 }
2091 }
2092 }
2093
2094 pub fn next_portable_change_frame(&mut self) -> Result<IOResult<Option<PortableChangeFrame>>> {
2102 self.file_size = self.file.size()? as usize;
2103 match return_if_io!(self.parse_next_portable_changes_frame()) {
2104 ParseResult::Frame(frame) => Ok(IOResult::Done(Some(PortableChangeFrame {
2105 end_offset: frame.end_offset as u64,
2106 commit_ts: frame.commit_ts,
2107 extension_record_count: frame.extension_record_count,
2108 payload: frame.portable_changes,
2109 }))),
2110 ParseResult::Eof | ParseResult::InvalidFrame => Ok(IOResult::Done(None)),
2111 }
2112 }
2113
2114 pub fn next_portable_changes(&mut self) -> Result<IOResult<Option<PortableChangeFrame>>> {
2120 loop {
2121 let Some(frame) = return_if_io!(self.next_portable_change_frame()) else {
2122 return Ok(IOResult::Done(None));
2123 };
2124 if !frame.payload.is_empty() {
2125 return Ok(IOResult::Done(Some(frame)));
2126 }
2127 }
2128 }
2129
2130 pub fn is_eof(&self) -> bool {
2131 self.remaining_bytes() == 0
2132 }
2133
2134 fn parse_decrypted_chunk_ops(
2138 plaintext: &[u8],
2139 start: usize,
2140 parsed_ops: &mut Vec<ParsedOp>,
2141 op_count: u32,
2142 commit_ts: u64,
2143 ) -> Result<usize> {
2144 let mut consumed = 0usize;
2145 while parsed_ops.len() < op_count as usize {
2146 match try_parse_one_op_from_buf(&plaintext[start + consumed..], commit_ts)? {
2147 Some((op, bytes_consumed)) => {
2148 consumed += bytes_consumed;
2149 parsed_ops.push(op);
2150 }
2151 None => break,
2152 }
2153 }
2154 Ok(consumed)
2155 }
2156
2157 fn carried_op_total_len_if_known(buf: &[u8]) -> Result<Option<usize>> {
2158 if buf.len() < 6 {
2161 return Ok(None);
2162 }
2163
2164 match buf[0] {
2165 OP_UPSERT_TABLE | OP_DELETE_TABLE | OP_UPSERT_INDEX | OP_DELETE_INDEX
2166 | OP_UPDATE_HEADER => {}
2167 tag => return Err(LimboError::Corrupt(format!("Unknown op tag: {tag}"))),
2168 }
2169
2170 let Some((payload_len_u64, varint_bytes)) = read_varint_partial(&buf[6..])? else {
2171 return Ok(None);
2173 };
2174 let payload_len = usize::try_from(payload_len_u64)
2175 .map_err(|_| LimboError::Corrupt("payload_len overflows usize".into()))?;
2176 let fixed = 6usize
2177 .checked_add(varint_bytes)
2178 .ok_or_else(|| LimboError::Corrupt("op header length overflow".into()))?;
2179 let total = fixed
2180 .checked_add(payload_len)
2181 .ok_or_else(|| LimboError::Corrupt("op payload length overflow".into()))?;
2182 Ok(Some(total))
2183 }
2184
2185 const MAX_SERIALIZED_OP_PREFIX_LEN: usize = 15;
2189
2190 fn read_and_decrypt_encrypted_chunk(
2192 &mut self,
2193 payload_ctx: &EncryptedPayloadReadContext,
2194 chunk_index: usize,
2195 running_crc: u32,
2196 ) -> Result<IOResult<EncryptedChunkReadResult>> {
2197 let plaintext_len = encrypted_chunk_plaintext_len(
2200 payload_ctx.payload_size,
2201 chunk_index,
2202 self.encrypted_payload_chunk_size,
2203 )?;
2204 let on_disk_size =
2205 encrypted_chunk_blob_size(plaintext_len, payload_ctx.tag_size, payload_ctx.nonce_size)?;
2206 let chunk_count = encrypted_payload_chunk_count(
2207 payload_ctx.payload_size,
2208 self.encrypted_payload_chunk_size,
2209 );
2210 let is_last_chunk = chunk_index + 1 == chunk_count;
2211
2212 let aad = build_encrypted_chunk_aad(
2213 payload_ctx.salt,
2214 is_last_chunk.then_some(payload_ctx.payload_size as u64),
2215 payload_ctx.op_count,
2216 payload_ctx.commit_ts,
2217 u32::try_from(chunk_index).map_err(|_| {
2218 LimboError::Corrupt("encrypted payload chunk index exceeds u32".to_string())
2219 })?,
2220 );
2221
2222 if self.remaining_bytes() < on_disk_size {
2223 return Ok(IOResult::Done(EncryptedChunkReadResult::Eof));
2224 }
2225 return_if_io!(self.read_more_data(on_disk_size));
2226 let start = self.buffer_offset;
2227 let end = start + on_disk_size;
2228
2229 let (next_crc, decrypted_plaintext_len) = {
2230 let encryption_ctx = self
2231 .encryption_ctx
2232 .as_ref()
2233 .expect("encryption_ctx must be set for encrypted payload");
2234 let decrypt_scratch = &mut self.decrypt_scratch;
2235 let buffer = self.buffer.read();
2236 let blob = &buffer[start..end];
2237 let next_crc = crc32c::crc32c_append(running_crc, blob);
2238 let ciphertext = &blob[..plaintext_len + payload_ctx.tag_size];
2239 let nonce = &blob[plaintext_len + payload_ctx.tag_size..];
2240 encryption_ctx
2241 .decrypt_chunk_into(ciphertext, nonce, &aad, decrypt_scratch)
2242 .map_err(|e| {
2243 LimboError::Corrupt(format!(
2244 "decrypt_chunk failed for chunk {chunk_index}: {e}"
2245 ))
2246 })?;
2247 (next_crc, decrypt_scratch.len())
2248 };
2249
2250 self.buffer_offset = end;
2251 if decrypted_plaintext_len != plaintext_len {
2252 return Err(LimboError::Corrupt(format!(
2253 "decrypted chunk length mismatch: expected {plaintext_len}, got {decrypted_plaintext_len}"
2254 )));
2255 }
2256
2257 Ok(IOResult::Done(EncryptedChunkReadResult::Ok {
2258 running_crc: next_crc,
2259 }))
2260 }
2261
2262 fn try_resolve_carried_encrypted_op_total_len(
2266 carry: &mut Vec<u8>,
2267 plaintext: &[u8],
2268 plaintext_start: &mut usize,
2269 ) -> Result<Option<usize>> {
2270 loop {
2271 if let Some(total_len) = Self::carried_op_total_len_if_known(carry)? {
2272 return Ok(Some(total_len));
2273 }
2274
2275 let available = plaintext.len().saturating_sub(*plaintext_start);
2276 if available == 0 {
2277 return Ok(None);
2279 }
2280
2281 if carry.len() >= Self::MAX_SERIALIZED_OP_PREFIX_LEN {
2282 return Err(LimboError::Corrupt(
2283 "carried encrypted op prefix could not resolve total length".into(),
2284 ));
2285 }
2286
2287 carry.push(plaintext[*plaintext_start]);
2288 *plaintext_start += 1;
2289 }
2290 }
2291
2292 fn try_finish_carried_encrypted_op(
2297 carry: &mut Vec<u8>,
2298 plaintext: &[u8],
2299 plaintext_start: &mut usize,
2300 parsed_ops: &mut Vec<ParsedOp>,
2301 op_count: u32,
2302 commit_ts: u64,
2303 ) -> Result<bool> {
2304 turso_assert!(!carry.is_empty());
2305 turso_assert!(parsed_ops.len() < op_count as usize);
2306
2307 let Some(carried_op_total_len) =
2309 Self::try_resolve_carried_encrypted_op_total_len(carry, plaintext, plaintext_start)?
2310 else {
2311 return Ok(false);
2312 };
2313
2314 if carry.len() > carried_op_total_len {
2317 return Err(LimboError::Corrupt(format!(
2318 "carried encrypted op exceeded computed length: len={} total={carried_op_total_len}",
2319 carry.len()
2320 )));
2321 }
2322 if carry.len() < carried_op_total_len {
2327 let available = plaintext.len().saturating_sub(*plaintext_start);
2328 if available == 0 {
2329 return Ok(false);
2330 }
2331 let take = (carried_op_total_len - carry.len()).min(available);
2332 carry.extend_from_slice(&plaintext[*plaintext_start..*plaintext_start + take]);
2333 *plaintext_start += take;
2334 if carry.len() < carried_op_total_len {
2335 return Ok(false);
2336 }
2337 }
2338
2339 turso_assert!(carry.len() == carried_op_total_len);
2341 match try_parse_one_op_from_buf(carry, commit_ts)? {
2342 Some((op, bytes_consumed)) if bytes_consumed == carry.len() => {
2343 parsed_ops.push(op);
2344 carry.clear();
2345 Ok(true)
2346 }
2347 Some((_, bytes_consumed)) => Err(LimboError::Corrupt(format!(
2348 "carried encrypted op consumed {bytes_consumed} bytes but carry holds {}",
2349 carry.len()
2350 ))),
2351 None => Err(LimboError::Corrupt(
2352 "carried encrypted op remained incomplete after reaching computed length".into(),
2353 )),
2354 }
2355 }
2356
2357 fn parse_encrypted_payload(
2362 &mut self,
2363 op_count: u32,
2364 payload_size: usize,
2365 commit_ts: u64,
2366 running_crc: u32,
2367 ) -> Result<IOResult<PayloadParseResult>> {
2368 let (nonce_size, tag_size) = {
2369 let enc = self
2370 .encryption_ctx
2371 .as_ref()
2372 .expect("encryption_ctx must be set for encrypted payload");
2373 (enc.nonce_size(), enc.tag_size())
2374 };
2375 let salt = self
2376 .header
2377 .as_ref()
2378 .expect("log header must be read before parsing")
2379 .salt;
2380 let payload_ctx = EncryptedPayloadReadContext {
2381 payload_size,
2382 op_count,
2383 commit_ts,
2384 salt,
2385 nonce_size,
2386 tag_size,
2387 };
2388 let mut running_crc = running_crc;
2389 let mut carry = Vec::with_capacity(self.encrypted_payload_chunk_size);
2394 let mut parsed_ops = Vec::with_capacity((op_count as usize).min(1024));
2397 let chunk_count =
2398 encrypted_payload_chunk_count(payload_size, self.encrypted_payload_chunk_size);
2399
2400 for chunk_index in 0..chunk_count {
2401 running_crc = match return_if_io!(self.read_and_decrypt_encrypted_chunk(
2403 &payload_ctx,
2404 chunk_index,
2405 running_crc,
2406 )) {
2407 EncryptedChunkReadResult::Ok { running_crc } => running_crc,
2408 EncryptedChunkReadResult::Eof => {
2409 return Ok(IOResult::Done(PayloadParseResult::Eof))
2410 }
2411 };
2412
2413 let mut plaintext_start = 0usize;
2414 let plaintext = self.decrypt_scratch.as_slice();
2415
2416 turso_assert!(
2417 parsed_ops.len() <= op_count as usize,
2418 "parsed_ops.len() exceeded declared op_count"
2419 );
2420 if !carry.is_empty() {
2421 if parsed_ops.len() == op_count as usize {
2422 return Err(LimboError::Corrupt(format!(
2423 "encrypted payload has trailing carried bytes after parsing all {op_count} ops"
2424 )));
2425 }
2426 match Self::try_finish_carried_encrypted_op(
2431 &mut carry,
2432 plaintext,
2433 &mut plaintext_start,
2434 &mut parsed_ops,
2435 op_count,
2436 commit_ts,
2437 ) {
2438 Ok(true) => {}
2439 Ok(false) => continue,
2440 Err(e) => {
2441 return Err(LimboError::Corrupt(format!(
2442 "encrypted carried-op parse error: {e}"
2443 )));
2444 }
2445 }
2446 }
2447 turso_assert!(
2449 carry.is_empty(),
2450 "carry must be empty before parsing fresh ops from the current decrypted chunk"
2451 );
2452
2453 let consumed = Self::parse_decrypted_chunk_ops(
2455 plaintext,
2456 plaintext_start,
2457 &mut parsed_ops,
2458 op_count,
2459 commit_ts,
2460 )?;
2461 plaintext_start += consumed;
2462 if plaintext_start < plaintext.len() {
2463 carry.extend_from_slice(&plaintext[plaintext_start..]);
2468 }
2469 }
2470
2471 if parsed_ops.len() != op_count as usize {
2473 return Err(LimboError::Corrupt(format!(
2474 "encrypted payload ended after {} parsed ops, expected {op_count}",
2475 parsed_ops.len()
2476 )));
2477 }
2478
2479 if !carry.is_empty() {
2481 return Err(LimboError::Corrupt(format!(
2482 "encrypted payload has {} trailing plaintext bytes after parsing all ops",
2483 carry.len()
2484 )));
2485 }
2486
2487 Ok(IOResult::Done(PayloadParseResult::Ok(
2488 parsed_ops,
2489 running_crc,
2490 )))
2491 }
2492
2493 fn read_encrypted_plaintext(
2494 &mut self,
2495 plaintext_size: usize,
2496 op_count: u32,
2497 commit_ts: u64,
2498 running_crc: u32,
2499 ) -> Result<IOResult<ReadEncryptedResult>> {
2500 let (nonce_size, tag_size) = {
2501 let enc = self
2502 .encryption_ctx
2503 .as_ref()
2504 .expect("encryption_ctx must be set for encrypted payload");
2505 (enc.nonce_size(), enc.tag_size())
2506 };
2507 let salt = self
2508 .header
2509 .as_ref()
2510 .expect("log header must be read before parsing")
2511 .salt;
2512 let payload_ctx = EncryptedPayloadReadContext {
2513 payload_size: plaintext_size,
2514 op_count,
2515 commit_ts,
2516 salt,
2517 nonce_size,
2518 tag_size,
2519 };
2520 let chunk_count =
2521 encrypted_payload_chunk_count(plaintext_size, self.encrypted_payload_chunk_size);
2522 let mut running_crc = running_crc;
2523 let mut plaintext = Vec::with_capacity(plaintext_size);
2524 for chunk_index in 0..chunk_count {
2525 running_crc = match return_if_io!(self.read_and_decrypt_encrypted_chunk(
2526 &payload_ctx,
2527 chunk_index,
2528 running_crc,
2529 )) {
2530 EncryptedChunkReadResult::Ok { running_crc } => running_crc,
2531 EncryptedChunkReadResult::Eof => return Ok(IOResult::Done(None)),
2532 };
2533 plaintext.extend_from_slice(&self.decrypt_scratch);
2534 }
2535 if plaintext.len() != plaintext_size {
2536 return Err(LimboError::Corrupt(format!(
2537 "encrypted plaintext size mismatch: expected {plaintext_size}, got {}",
2538 plaintext.len()
2539 )));
2540 }
2541 Ok(IOResult::Done(Some((plaintext, running_crc))))
2542 }
2543
2544 fn parse_streaming_payload(&mut self) -> Result<IOResult<PayloadOutcome>> {
2553 loop {
2554 let (op_index, op_count, commit_ts) = {
2555 let fip = self
2556 .frame_in_progress
2557 .as_ref()
2558 .expect("frame in progress while parsing streaming payload");
2559 (fip.op_index, fip.op_count, fip.commit_ts)
2560 };
2561
2562 if op_index >= op_count {
2563 let (payload_size, payload_bytes_read) = {
2564 let fip = self
2565 .frame_in_progress
2566 .as_ref()
2567 .expect("frame in progress while parsing streaming payload");
2568 (fip.payload_size, fip.payload_bytes_read)
2569 };
2570 if payload_size as u64 != payload_bytes_read {
2571 return Err(LimboError::Corrupt(format!(
2572 "payload_size ({payload_size}) != payload_bytes_read ({payload_bytes_read})"
2573 )));
2574 }
2575 return Ok(IOResult::Done(PayloadOutcome::Ok));
2576 }
2577
2578 let mut running_crc = self
2581 .frame_in_progress
2582 .as_ref()
2583 .expect("frame in progress while parsing streaming payload")
2584 .running_crc;
2585 let mut payload_bytes_read = self
2586 .frame_in_progress
2587 .as_ref()
2588 .expect("frame in progress while parsing streaming payload")
2589 .payload_bytes_read;
2590
2591 let op_bytes = match return_if_io!(self.try_consume_fixed::<6>()) {
2593 Some(bytes) => bytes,
2594 None => return Ok(IOResult::Done(PayloadOutcome::Eof)),
2595 };
2596 running_crc = crc32c::crc32c_append(running_crc, &op_bytes);
2597 let tag = op_bytes[0];
2598 let flags = op_bytes[1];
2599 let table_id_i32 =
2600 i32::from_le_bytes([op_bytes[2], op_bytes[3], op_bytes[4], op_bytes[5]]);
2601 let table_id = match tag {
2602 OP_UPSERT_TABLE | OP_DELETE_TABLE | OP_UPSERT_INDEX | OP_DELETE_INDEX => {
2603 if flags & !OP_ALLOWED_FLAGS != 0 || table_id_i32 >= 0 {
2604 return Err(LimboError::Corrupt(format!(
2605 "invalid op flags={flags:#x} or table_id={table_id_i32} for tag={tag}"
2606 )));
2607 }
2608 Some(MVTableId::from(table_id_i32 as i64))
2609 }
2610 OP_UPDATE_HEADER => {
2611 if flags != 0 || table_id_i32 != 0 {
2612 return Err(LimboError::Corrupt(format!(
2613 "OP_UPDATE_HEADER has non-zero flags={flags:#x} or table_id={table_id_i32}"
2614 )));
2615 }
2616 None
2617 }
2618 _ => {
2619 return Err(LimboError::Corrupt(format!("unknown op tag {tag}")));
2620 }
2621 };
2622 let btree_resident = (flags & OP_FLAG_BTREE_RESIDENT) != 0;
2623 let has_portable_extension = (flags & OP_FLAG_PORTABLE_EXTENSION) != 0;
2624
2625 let (payload_len, payload_len_bytes, payload_len_bytes_len) =
2626 match return_if_io!(self.consume_varint_bytes()) {
2627 Some((value, bytes, len)) => (value, bytes, len),
2628 None => return Ok(IOResult::Done(PayloadOutcome::Eof)),
2629 };
2630 running_crc =
2631 crc32c::crc32c_append(running_crc, &payload_len_bytes[..payload_len_bytes_len]);
2632 let payload_len = usize::try_from(payload_len)
2633 .map_err(|e| LimboError::Corrupt(format!("payload_len overflows usize: {e}")))?;
2634
2635 let payload = match return_if_io!(self.try_consume_bytes(payload_len)) {
2636 Some(bytes) => bytes,
2637 None => return Ok(IOResult::Done(PayloadOutcome::Eof)),
2638 };
2639 running_crc = crc32c::crc32c_append(running_crc, &payload);
2640
2641 let (portable_extension, extension_total_bytes) = if has_portable_extension {
2642 let (extension_len, extension_len_bytes, extension_len_bytes_len) =
2643 match return_if_io!(self.consume_varint_bytes()) {
2644 Some((value, bytes, len)) => (value, bytes, len),
2645 None => return Ok(IOResult::Done(PayloadOutcome::Eof)),
2646 };
2647 running_crc = crc32c::crc32c_append(
2648 running_crc,
2649 &extension_len_bytes[..extension_len_bytes_len],
2650 );
2651 let extension_len = usize::try_from(extension_len).map_err(|e| {
2652 LimboError::Corrupt(format!("op extension length overflows usize: {e}"))
2653 })?;
2654 let extension = match return_if_io!(self.try_consume_bytes(extension_len)) {
2655 Some(bytes) => bytes,
2656 None => return Ok(IOResult::Done(PayloadOutcome::Eof)),
2657 };
2658 running_crc = crc32c::crc32c_append(running_crc, &extension);
2659 (extension, extension_len_bytes_len + extension_len)
2660 } else {
2661 (Vec::new(), 0)
2662 };
2663
2664 let op_total_bytes = 6 + payload_len_bytes_len + payload_len + extension_total_bytes;
2665 payload_bytes_read = u64::try_from(op_total_bytes)
2666 .ok()
2667 .and_then(|op_size| payload_bytes_read.checked_add(op_size))
2668 .ok_or_else(|| LimboError::Corrupt("payload_bytes_read overflow".to_string()))?;
2669
2670 let parsed_op = match tag {
2671 OP_UPSERT_TABLE => {
2672 let table_id = table_id.expect("table op must carry table id");
2673 let (rowid_u64, rowid_len) = read_varint(&payload).map_err(|e| {
2674 LimboError::Corrupt(format!(
2675 "failed to read rowid varint in upsert op: {e}"
2676 ))
2677 })?;
2678 let rowid_i64 = rowid_u64 as i64;
2679 if rowid_len > payload.len() {
2680 return Err(LimboError::Corrupt(
2681 "upsert op rowid varint extends beyond payload".to_string(),
2682 ));
2683 }
2684 let mut payload = payload;
2685 let record_bytes = payload.split_off(rowid_len);
2686 let rowid = RowID::new(table_id, RowKey::Int(rowid_i64));
2687 ParsedOp::UpsertTable {
2688 table_id,
2689 rowid,
2690 record_bytes,
2691 commit_ts,
2692 btree_resident,
2693 }
2694 }
2695 OP_DELETE_TABLE => {
2696 let table_id = table_id.expect("table op must carry table id");
2697 let (rowid_u64, rowid_len) = read_varint(&payload).map_err(|e| {
2698 LimboError::Corrupt(format!(
2699 "failed to read rowid varint in delete op: {e}"
2700 ))
2701 })?;
2702 if rowid_len > payload.len() {
2703 return Err(LimboError::Corrupt(format!(
2704 "delete op rowid varint len {rowid_len} > payload len {}",
2705 payload.len()
2706 )));
2707 }
2708 let rowid_i64 = rowid_u64 as i64;
2709 let mut payload = payload;
2710 let mut record_bytes = payload.split_off(rowid_len);
2711 let mut pk_record_bytes = Vec::new();
2712 if !portable_extension.is_empty() {
2713 let decoded = decode_delete_portable_extension(&portable_extension)?;
2714 if record_bytes.is_empty() {
2715 record_bytes = decoded.identity_record;
2716 }
2717 pk_record_bytes = decoded.pk_record;
2718 }
2719 let rowid = RowID::new(table_id, RowKey::Int(rowid_i64));
2720 ParsedOp::DeleteTable {
2721 rowid,
2722 record_bytes,
2723 pk_record_bytes,
2724 commit_ts,
2725 btree_resident,
2726 }
2727 }
2728 OP_UPSERT_INDEX => {
2729 let table_id = table_id.expect("index op must carry table id");
2730 ParsedOp::UpsertIndex {
2731 table_id,
2732 payload,
2733 commit_ts,
2734 btree_resident,
2735 }
2736 }
2737 OP_DELETE_INDEX => {
2738 let table_id = table_id.expect("index op must carry table id");
2739 ParsedOp::DeleteIndex {
2740 table_id,
2741 payload,
2742 commit_ts,
2743 btree_resident,
2744 }
2745 }
2746 OP_UPDATE_HEADER => {
2747 if payload.len() != DatabaseHeader::SIZE {
2748 return Err(LimboError::Corrupt(format!(
2749 "OP_UPDATE_HEADER payload len {} != DatabaseHeader::SIZE {}",
2750 payload.len(),
2751 DatabaseHeader::SIZE
2752 )));
2753 }
2754 let mut bytes = [0u8; DatabaseHeader::SIZE];
2755 bytes.copy_from_slice(&payload);
2756 let header = *bytemuck::from_bytes::<DatabaseHeader>(&bytes);
2757 if header.magic != *b"SQLite format 3\0" {
2758 return Err(LimboError::Corrupt(
2759 "OP_UPDATE_HEADER has invalid SQLite magic".to_string(),
2760 ));
2761 }
2762 ParsedOp::UpdateHeader { header, commit_ts }
2763 }
2764 _ => {
2765 return Err(LimboError::Corrupt(format!(
2766 "unknown op tag {tag} in payload"
2767 )));
2768 }
2769 };
2770
2771 {
2774 let fip = self
2775 .frame_in_progress
2776 .as_mut()
2777 .expect("frame in progress while parsing streaming payload");
2778 fip.parsed_ops.push(parsed_op);
2779 fip.running_crc = running_crc;
2780 fip.payload_bytes_read = payload_bytes_read;
2781 fip.op_index += 1;
2782 }
2783 self.advance_checkpoint();
2784 }
2785 }
2786
2787 fn parse_next_transaction(&mut self) -> Result<IOResult<ParseResult>> {
2799 loop {
2800 if self.frame_in_progress.is_none() {
2801 self.buffer_offset = self.frame_anchor;
2803 if self.remaining_bytes() < self.tx_min_frame_size() {
2804 return Ok(IOResult::Done(ParseResult::Eof));
2805 }
2806 let frame_start = self.offset.saturating_sub(self.bytes_can_read());
2807 self.frame_in_progress = Some(FrameInProgress::new(frame_start));
2808 } else {
2809 self.buffer_offset = self.frame_anchor;
2812 }
2813
2814 let phase = self
2815 .frame_in_progress
2816 .as_ref()
2817 .expect("frame in progress")
2818 .phase;
2819 match phase {
2820 FramePhase::Header => {
2821 let header = match return_if_io!(self.parse_frame_header()) {
2822 HeaderParseOutcome::Ok(header) => header,
2823 HeaderParseOutcome::Eof => return self.abort_frame_eof(),
2824 HeaderParseOutcome::Invalid => return self.invalidate_frame(),
2825 };
2826 let next_phase = if self.encryption_ctx.is_none() && header.extension_size > 0 {
2830 FramePhase::ExtensionBlock
2831 } else {
2832 FramePhase::Payload
2833 };
2834 {
2835 let fip = self.frame_in_progress.as_mut().expect("frame in progress");
2836 fip.payload_size = header.payload_size;
2837 fip.op_count = header.op_count;
2838 fip.commit_ts = header.commit_ts;
2839 fip.extension_size = header.extension_size;
2840 fip.extension_record_count = header.extension_record_count;
2841 fip.frame_flags = header.frame_flags;
2842 fip.running_crc = header.running_crc;
2843 fip.phase = next_phase;
2844 }
2845 self.advance_checkpoint();
2846 }
2847 FramePhase::ExtensionBlock => {
2848 let (extension_size, extension_record_count, running_crc) = {
2849 let fip = self.frame_in_progress.as_ref().expect("frame in progress");
2850 (
2851 fip.extension_size,
2852 fip.extension_record_count,
2853 fip.running_crc,
2854 )
2855 };
2856 let bytes = match return_if_io!(self.try_consume_bytes(extension_size)) {
2857 Some(bytes) => bytes,
2858 None => return self.abort_frame_eof(),
2859 };
2860 let running_crc = crc32c::crc32c_append(running_crc, &bytes);
2861 let portable_changes = match find_extension_payload(
2862 &bytes,
2863 extension_record_count,
2864 EXTENSION_TYPE_PORTABLE_CHANGES,
2865 ) {
2866 Ok(payload) => payload,
2867 Err(LimboError::Corrupt(msg)) => {
2868 tracing::warn!("corrupt extension block: {msg}");
2869 return self.invalidate_frame();
2870 }
2871 Err(e) => return Err(e),
2872 };
2873 {
2874 let fip = self.frame_in_progress.as_mut().expect("frame in progress");
2875 fip.portable_changes = portable_changes;
2876 fip.running_crc = running_crc;
2877 fip.phase = FramePhase::Payload;
2878 }
2879 self.advance_checkpoint();
2880 }
2881 FramePhase::Payload => match self.parse_payload_phase() {
2882 Ok(IOResult::Done(PayloadOutcome::Ok)) => {
2883 self.frame_in_progress
2884 .as_mut()
2885 .expect("frame in progress")
2886 .phase = FramePhase::Trailer;
2887 self.advance_checkpoint();
2888 }
2889 Ok(IOResult::Done(PayloadOutcome::Eof)) => return self.abort_frame_eof(),
2890 Ok(IOResult::IO(io)) => return Ok(IOResult::IO(io)),
2891 Err(LimboError::Corrupt(msg)) => {
2892 tracing::warn!("corrupt payload: {msg}");
2893 return self.invalidate_frame();
2894 }
2895 Err(e) => return Err(e),
2896 },
2897 FramePhase::Trailer => {
2898 let trailer_bytes =
2900 match return_if_io!(self.try_consume_fixed::<TX_TRAILER_SIZE>()) {
2901 Some(bytes) => bytes,
2902 None => return self.abort_frame_eof(),
2903 };
2904 let crc32c_expected = u32::from_le_bytes([
2905 trailer_bytes[0],
2906 trailer_bytes[1],
2907 trailer_bytes[2],
2908 trailer_bytes[3],
2909 ]);
2910 let end_magic = u32::from_le_bytes([
2911 trailer_bytes[4],
2912 trailer_bytes[5],
2913 trailer_bytes[6],
2914 trailer_bytes[7],
2915 ]);
2916 let running_crc = self
2917 .frame_in_progress
2918 .as_ref()
2919 .expect("frame in progress")
2920 .running_crc;
2921 if crc32c_expected != running_crc {
2922 return self.invalidate_frame();
2923 }
2924 if end_magic != END_MAGIC {
2925 return self.invalidate_frame();
2926 }
2927 return self.commit_frame();
2928 }
2929 }
2930 }
2931 }
2932
2933 fn parse_frame_header(&mut self) -> Result<IOResult<HeaderParseOutcome>> {
2940 let mut header_bytes = match return_if_io!(self.try_consume_bytes(TX_HEADER_SIZE)) {
2946 Some(bytes) => bytes,
2947 None => return Ok(IOResult::Done(HeaderParseOutcome::Eof)),
2948 };
2949
2950 let frame_magic = u32::from_le_bytes([
2951 header_bytes[0],
2952 header_bytes[1],
2953 header_bytes[2],
2954 header_bytes[3],
2955 ]);
2956 let is_v2 = self
2957 .header
2958 .as_ref()
2959 .is_some_and(|header| header.version == LOG_VERSION_V2);
2960 let has_extension_header = !is_v2 && frame_magic == EXT_FRAME_MAGIC;
2961 if frame_magic != FRAME_MAGIC && !has_extension_header {
2962 return Ok(IOResult::Done(HeaderParseOutcome::Invalid));
2963 }
2964 if is_v2 && frame_magic != FRAME_MAGIC {
2965 return Ok(IOResult::Done(HeaderParseOutcome::Invalid));
2966 }
2967 if has_extension_header {
2968 let Some(extension_header) =
2969 return_if_io!(self.try_consume_bytes(TX_EXT_HEADER_SIZE - TX_HEADER_SIZE))
2970 else {
2971 return Ok(IOResult::Done(HeaderParseOutcome::Eof));
2972 };
2973 header_bytes.extend_from_slice(&extension_header);
2974 }
2975 let payload_size_u64 = u64::from_le_bytes([
2976 header_bytes[4],
2977 header_bytes[5],
2978 header_bytes[6],
2979 header_bytes[7],
2980 header_bytes[8],
2981 header_bytes[9],
2982 header_bytes[10],
2983 header_bytes[11],
2984 ]);
2985 let op_count = u32::from_le_bytes([
2986 header_bytes[12],
2987 header_bytes[13],
2988 header_bytes[14],
2989 header_bytes[15],
2990 ]);
2991 let commit_ts = u64::from_le_bytes([
2992 header_bytes[16],
2993 header_bytes[17],
2994 header_bytes[18],
2995 header_bytes[19],
2996 header_bytes[20],
2997 header_bytes[21],
2998 header_bytes[22],
2999 header_bytes[23],
3000 ]);
3001 let (extension_size, extension_record_count, frame_flags) = if has_extension_header {
3002 let extension_size_u64 = u64::from_le_bytes([
3003 header_bytes[24],
3004 header_bytes[25],
3005 header_bytes[26],
3006 header_bytes[27],
3007 header_bytes[28],
3008 header_bytes[29],
3009 header_bytes[30],
3010 header_bytes[31],
3011 ]);
3012 let extension_size = match usize::try_from(extension_size_u64) {
3013 Ok(v) => v,
3014 Err(e) => {
3015 tracing::warn!("extension_size overflows usize: {e}");
3016 return Ok(IOResult::Done(HeaderParseOutcome::Invalid));
3017 }
3018 };
3019 let extension_record_count = u32::from_le_bytes([
3020 header_bytes[32],
3021 header_bytes[33],
3022 header_bytes[34],
3023 header_bytes[35],
3024 ]);
3025 let frame_flags = u32::from_le_bytes([
3026 header_bytes[36],
3027 header_bytes[37],
3028 header_bytes[38],
3029 header_bytes[39],
3030 ]);
3031 if frame_flags & !TX_FRAME_FLAG_HAS_EXTENSION_BLOCK != 0 {
3032 return Ok(IOResult::Done(HeaderParseOutcome::Invalid));
3033 }
3034 if extension_size == 0 && extension_record_count != 0 {
3035 return Ok(IOResult::Done(HeaderParseOutcome::Invalid));
3036 }
3037 if extension_size > 0 && frame_flags & TX_FRAME_FLAG_HAS_EXTENSION_BLOCK == 0 {
3038 return Ok(IOResult::Done(HeaderParseOutcome::Invalid));
3039 }
3040 (extension_size, extension_record_count, frame_flags)
3041 } else {
3042 (0, 0, 0)
3043 };
3044
3045 let payload_size = match usize::try_from(payload_size_u64) {
3046 Ok(v) => v,
3047 Err(e) => {
3048 tracing::warn!("payload_size overflows usize: {e}");
3049 return Ok(IOResult::Done(HeaderParseOutcome::Invalid));
3050 }
3051 };
3052
3053 let running_crc = crc32c::crc32c_append(self.running_crc, &header_bytes);
3055
3056 Ok(IOResult::Done(HeaderParseOutcome::Ok(FrameHeader {
3057 payload_size,
3058 op_count,
3059 commit_ts,
3060 extension_size,
3061 extension_record_count,
3062 frame_flags,
3063 running_crc,
3064 })))
3065 }
3066
3067 fn parse_payload_phase(&mut self) -> Result<IOResult<PayloadOutcome>> {
3075 let (payload_size, op_count, commit_ts, extension_size, extension_record_count, header_crc) = {
3076 let fip = self.frame_in_progress.as_ref().expect("frame in progress");
3077 (
3078 fip.payload_size,
3079 fip.op_count,
3080 fip.commit_ts,
3081 fip.extension_size,
3082 fip.extension_record_count,
3083 fip.running_crc,
3084 )
3085 };
3086 let encrypted_extension_size = if self.encryption_ctx.is_some() {
3087 extension_size
3088 } else {
3089 0
3090 };
3091
3092 if encrypted_extension_size > 0 {
3093 let plaintext_size = payload_size
3094 .checked_add(encrypted_extension_size)
3095 .ok_or_else(|| {
3096 LimboError::Corrupt("encrypted plaintext size overflows usize".into())
3097 })?;
3098 let Some((plaintext, running_crc)) = return_if_io!(self.read_encrypted_plaintext(
3099 plaintext_size,
3100 op_count,
3101 commit_ts,
3102 header_crc,
3103 )) else {
3104 return Ok(IOResult::Done(PayloadOutcome::Eof));
3105 };
3106 let recovery_start = extension_size;
3107 let recovery_end = recovery_start
3108 .checked_add(payload_size)
3109 .ok_or_else(|| LimboError::Corrupt("recovery payload offset overflow".into()))?;
3110 let portable_changes = find_extension_payload(
3111 &plaintext[..extension_size],
3112 extension_record_count,
3113 EXTENSION_TYPE_PORTABLE_CHANGES,
3114 )?;
3115 let parsed_ops = parse_ops_from_plaintext(
3116 &plaintext[recovery_start..recovery_end],
3117 payload_size,
3118 op_count,
3119 commit_ts,
3120 )?;
3121 let fip = self.frame_in_progress.as_mut().expect("frame in progress");
3122 fip.parsed_ops = parsed_ops;
3123 fip.portable_changes = portable_changes;
3124 fip.running_crc = running_crc;
3125 return Ok(IOResult::Done(PayloadOutcome::Ok));
3126 }
3127
3128 if self.encryption_ctx.is_some() {
3129 let (parsed_ops, running_crc) = match self.parse_encrypted_payload(
3130 op_count,
3131 payload_size,
3132 commit_ts,
3133 header_crc,
3134 )? {
3135 IOResult::Done(PayloadParseResult::Ok(ops, crc)) => (ops, crc),
3136 IOResult::Done(PayloadParseResult::Eof) => {
3137 return Ok(IOResult::Done(PayloadOutcome::Eof))
3138 }
3139 IOResult::IO(io) => return Ok(IOResult::IO(io)),
3140 };
3141 let fip = self.frame_in_progress.as_mut().expect("frame in progress");
3142 fip.parsed_ops = parsed_ops;
3143 fip.running_crc = running_crc;
3144 return Ok(IOResult::Done(PayloadOutcome::Ok));
3145 }
3146
3147 self.parse_streaming_payload()
3150 }
3151
3152 fn advance_checkpoint(&mut self) {
3157 self.frame_anchor = self.buffer_offset;
3158 }
3159
3160 fn abort_frame_eof(&mut self) -> Result<IOResult<ParseResult>> {
3165 self.frame_in_progress = None;
3166 Ok(IOResult::Done(ParseResult::Eof))
3167 }
3168
3169 fn invalidate_frame(&mut self) -> Result<IOResult<ParseResult>> {
3174 let frame_start = self
3175 .frame_in_progress
3176 .as_ref()
3177 .expect("frame in progress")
3178 .frame_start;
3179 self.last_valid_offset = frame_start;
3180 self.frame_in_progress = None;
3181 Ok(IOResult::Done(ParseResult::InvalidFrame))
3182 }
3183
3184 fn commit_frame(&mut self) -> Result<IOResult<ParseResult>> {
3188 let fip = self.frame_in_progress.take().expect("frame in progress");
3189 self.last_valid_offset = self.offset.saturating_sub(self.bytes_can_read());
3190 self.running_crc = fip.running_crc;
3191 self.frame_anchor = self.buffer_offset;
3192 Ok(IOResult::Done(ParseResult::Frame(ParsedFrame {
3193 ops: fip.parsed_ops,
3194 portable_changes: fip.portable_changes,
3195 extension_record_count: fip.extension_record_count,
3196 frame_flags: fip.frame_flags,
3197 commit_ts: fip.commit_ts,
3198 end_offset: self.last_valid_offset,
3199 })))
3200 }
3201
3202 fn consume_and_crc_bytes(
3203 &mut self,
3204 mut amount: usize,
3205 mut running_crc: u32,
3206 ) -> Result<IOResult<Option<u32>>> {
3207 const CHUNK_SIZE: usize = 64 * 1024;
3208 while amount > 0 {
3209 let chunk_len = amount.min(CHUNK_SIZE);
3210 let Some(bytes) = return_if_io!(self.try_consume_bytes(chunk_len)) else {
3211 return Ok(IOResult::Done(None));
3212 };
3213 running_crc = crc32c::crc32c_append(running_crc, &bytes);
3214 amount -= chunk_len;
3215 }
3216 Ok(IOResult::Done(Some(running_crc)))
3217 }
3218
3219 fn encrypted_payload_on_disk_size(&self, payload_size: usize) -> Result<usize> {
3220 let Some(encryption_ctx) = self.encryption_ctx.as_ref() else {
3221 return Ok(payload_size);
3222 };
3223 let mut on_disk_size = 0usize;
3224 for chunk_index in
3225 0..encrypted_payload_chunk_count(payload_size, self.encrypted_payload_chunk_size)
3226 {
3227 let plaintext_len = encrypted_chunk_plaintext_len(
3228 payload_size,
3229 chunk_index,
3230 self.encrypted_payload_chunk_size,
3231 )?;
3232 on_disk_size = on_disk_size
3233 .checked_add(encrypted_chunk_blob_size(
3234 plaintext_len,
3235 encryption_ctx.tag_size(),
3236 encryption_ctx.nonce_size(),
3237 )?)
3238 .ok_or_else(|| {
3239 LimboError::Corrupt("encrypted payload size overflows usize".to_string())
3240 })?;
3241 }
3242 Ok(on_disk_size)
3243 }
3244
3245 fn parse_next_portable_changes_frame(&mut self) -> Result<IOResult<ParseResult>> {
3246 self.buffer_offset = self.frame_anchor;
3249 if self
3250 .header
3251 .as_ref()
3252 .is_some_and(|h| h.version == LOG_VERSION_V2)
3253 {
3254 return Ok(IOResult::Done(ParseResult::Eof));
3255 }
3256 if self.remaining_bytes() < TX_MIN_FRAME_SIZE {
3257 return Ok(IOResult::Done(ParseResult::Eof));
3258 }
3259 let frame_start = self.offset.saturating_sub(self.bytes_can_read());
3260
3261 let mut header_bytes = match return_if_io!(self.try_consume_bytes(TX_HEADER_SIZE)) {
3262 Some(bytes) => bytes,
3263 None => return Ok(IOResult::Done(ParseResult::Eof)),
3264 };
3265
3266 let frame_magic = u32::from_le_bytes([
3267 header_bytes[0],
3268 header_bytes[1],
3269 header_bytes[2],
3270 header_bytes[3],
3271 ]);
3272 let has_extension_header = frame_magic == EXT_FRAME_MAGIC;
3273 if frame_magic != FRAME_MAGIC && !has_extension_header {
3274 self.last_valid_offset = frame_start;
3275 return Ok(IOResult::Done(ParseResult::InvalidFrame));
3276 }
3277 if has_extension_header {
3278 let Some(extension_header) =
3279 return_if_io!(self.try_consume_bytes(TX_EXT_HEADER_SIZE - TX_HEADER_SIZE))
3280 else {
3281 return Ok(IOResult::Done(ParseResult::Eof));
3282 };
3283 header_bytes.extend_from_slice(&extension_header);
3284 }
3285 let payload_size_u64 = u64::from_le_bytes([
3286 header_bytes[4],
3287 header_bytes[5],
3288 header_bytes[6],
3289 header_bytes[7],
3290 header_bytes[8],
3291 header_bytes[9],
3292 header_bytes[10],
3293 header_bytes[11],
3294 ]);
3295 let op_count = u32::from_le_bytes([
3296 header_bytes[12],
3297 header_bytes[13],
3298 header_bytes[14],
3299 header_bytes[15],
3300 ]);
3301 let commit_ts = u64::from_le_bytes([
3302 header_bytes[16],
3303 header_bytes[17],
3304 header_bytes[18],
3305 header_bytes[19],
3306 header_bytes[20],
3307 header_bytes[21],
3308 header_bytes[22],
3309 header_bytes[23],
3310 ]);
3311 let (extension_size_u64, extension_record_count, frame_flags) = if has_extension_header {
3312 let extension_size_u64 = u64::from_le_bytes([
3313 header_bytes[24],
3314 header_bytes[25],
3315 header_bytes[26],
3316 header_bytes[27],
3317 header_bytes[28],
3318 header_bytes[29],
3319 header_bytes[30],
3320 header_bytes[31],
3321 ]);
3322 let extension_record_count = u32::from_le_bytes([
3323 header_bytes[32],
3324 header_bytes[33],
3325 header_bytes[34],
3326 header_bytes[35],
3327 ]);
3328 let frame_flags = u32::from_le_bytes([
3329 header_bytes[36],
3330 header_bytes[37],
3331 header_bytes[38],
3332 header_bytes[39],
3333 ]);
3334 if frame_flags & !TX_FRAME_FLAG_HAS_EXTENSION_BLOCK != 0 {
3335 self.last_valid_offset = frame_start;
3336 return Ok(IOResult::Done(ParseResult::InvalidFrame));
3337 }
3338 if extension_size_u64 == 0 && extension_record_count != 0 {
3339 self.last_valid_offset = frame_start;
3340 return Ok(IOResult::Done(ParseResult::InvalidFrame));
3341 }
3342 if extension_size_u64 > 0 && frame_flags & TX_FRAME_FLAG_HAS_EXTENSION_BLOCK == 0 {
3343 self.last_valid_offset = frame_start;
3344 return Ok(IOResult::Done(ParseResult::InvalidFrame));
3345 }
3346 (extension_size_u64, extension_record_count, frame_flags)
3347 } else {
3348 (0, 0, 0)
3349 };
3350
3351 let payload_size = match usize::try_from(payload_size_u64) {
3352 Ok(v) => v,
3353 Err(e) => {
3354 tracing::warn!("payload_size overflows usize: {e}");
3355 self.last_valid_offset = frame_start;
3356 return Ok(IOResult::Done(ParseResult::InvalidFrame));
3357 }
3358 };
3359 let extension_size = match usize::try_from(extension_size_u64) {
3360 Ok(v) => v,
3361 Err(e) => {
3362 tracing::warn!("extension_size overflows usize: {e}");
3363 self.last_valid_offset = frame_start;
3364 return Ok(IOResult::Done(ParseResult::InvalidFrame));
3365 }
3366 };
3367
3368 let running_crc = crc32c::crc32c_append(self.running_crc, &header_bytes);
3369 let encrypted_extension_size = if self.encryption_ctx.is_some() {
3370 extension_size
3371 } else {
3372 0
3373 };
3374 let payload_on_disk_size = match self.encrypted_payload_on_disk_size(
3375 payload_size
3376 .checked_add(encrypted_extension_size)
3377 .ok_or_else(|| {
3378 LimboError::Corrupt(
3379 "payload plus encrypted extension size overflows usize".to_string(),
3380 )
3381 })?,
3382 ) {
3383 Ok(size) => size,
3384 Err(LimboError::Corrupt(msg)) => {
3385 tracing::warn!("corrupt payload size: {msg}");
3386 self.last_valid_offset = frame_start;
3387 return Ok(IOResult::Done(ParseResult::InvalidFrame));
3388 }
3389 Err(e) => return Err(e),
3390 };
3391 let (portable_changes, running_crc) = if encrypted_extension_size > 0 {
3392 let plaintext_size = payload_size
3393 .checked_add(encrypted_extension_size)
3394 .ok_or_else(|| {
3395 LimboError::Corrupt("encrypted plaintext size overflows usize".into())
3396 })?;
3397 let Some((plaintext, running_crc)) = return_if_io!(self.read_encrypted_plaintext(
3398 plaintext_size,
3399 op_count,
3400 commit_ts,
3401 running_crc,
3402 )) else {
3403 return Ok(IOResult::Done(ParseResult::Eof));
3404 };
3405 let portable_changes = match find_extension_payload(
3406 &plaintext[..extension_size],
3407 extension_record_count,
3408 EXTENSION_TYPE_PORTABLE_CHANGES,
3409 ) {
3410 Ok(payload) => payload,
3411 Err(LimboError::Corrupt(msg)) => {
3412 tracing::warn!("corrupt extension block: {msg}");
3413 self.last_valid_offset = frame_start;
3414 return Ok(IOResult::Done(ParseResult::InvalidFrame));
3415 }
3416 Err(e) => return Err(e),
3417 };
3418 (portable_changes, running_crc)
3419 } else {
3420 let (portable_changes, running_crc) = if extension_size > 0 {
3421 match return_if_io!(self.try_consume_bytes(extension_size)) {
3422 Some(bytes) => {
3423 let running_crc = crc32c::crc32c_append(running_crc, &bytes);
3424 let portable_changes = match find_extension_payload(
3425 &bytes,
3426 extension_record_count,
3427 EXTENSION_TYPE_PORTABLE_CHANGES,
3428 ) {
3429 Ok(payload) => payload,
3430 Err(LimboError::Corrupt(msg)) => {
3431 tracing::warn!("corrupt extension block: {msg}");
3432 self.last_valid_offset = frame_start;
3433 return Ok(IOResult::Done(ParseResult::InvalidFrame));
3434 }
3435 Err(e) => return Err(e),
3436 };
3437 (portable_changes, running_crc)
3438 }
3439 None => return Ok(IOResult::Done(ParseResult::Eof)),
3440 }
3441 } else {
3442 (Vec::new(), running_crc)
3443 };
3444 let Some(running_crc) =
3445 return_if_io!(self.consume_and_crc_bytes(payload_on_disk_size, running_crc))
3446 else {
3447 return Ok(IOResult::Done(ParseResult::Eof));
3448 };
3449 (portable_changes, running_crc)
3450 };
3451
3452 let trailer_bytes = match return_if_io!(self.try_consume_fixed::<TX_TRAILER_SIZE>()) {
3453 Some(bytes) => bytes,
3454 None => return Ok(IOResult::Done(ParseResult::Eof)),
3455 };
3456 let crc32c_expected = u32::from_le_bytes([
3457 trailer_bytes[0],
3458 trailer_bytes[1],
3459 trailer_bytes[2],
3460 trailer_bytes[3],
3461 ]);
3462 let end_magic = u32::from_le_bytes([
3463 trailer_bytes[4],
3464 trailer_bytes[5],
3465 trailer_bytes[6],
3466 trailer_bytes[7],
3467 ]);
3468 if crc32c_expected != running_crc {
3469 self.last_valid_offset = frame_start;
3470 return Ok(IOResult::Done(ParseResult::InvalidFrame));
3471 }
3472 if end_magic != END_MAGIC {
3473 self.last_valid_offset = frame_start;
3474 return Ok(IOResult::Done(ParseResult::InvalidFrame));
3475 }
3476
3477 self.last_valid_offset = self.offset.saturating_sub(self.bytes_can_read());
3478 self.running_crc = running_crc;
3479 self.frame_anchor = self.buffer_offset;
3480 Ok(IOResult::Done(ParseResult::Frame(ParsedFrame {
3481 ops: Vec::new(),
3482 portable_changes,
3483 extension_record_count,
3484 frame_flags,
3485 commit_ts,
3486 end_offset: self.last_valid_offset,
3487 })))
3488 }
3489
3490 pub(crate) fn parsed_op_to_streaming(
3491 &self,
3492 parsed_op: ParsedOp,
3493 get_index_info: &mut impl FnMut(MVTableId, IndexOpKind) -> Result<Arc<IndexInfo>>,
3494 ) -> Result<StreamingResult> {
3495 self.parsed_op_to_streaming_in(parsed_op, get_index_info, TursoAllocator)
3496 }
3497
3498 pub(crate) fn parsed_op_to_streaming_in<A: ConcurrentAllocator>(
3499 &self,
3500 parsed_op: ParsedOp,
3501 get_index_info: &mut impl FnMut(MVTableId, IndexOpKind) -> Result<Arc<IndexInfo>>,
3502 alloc: A,
3503 ) -> Result<StreamingResult> {
3504 match parsed_op {
3505 ParsedOp::UpsertTable {
3506 table_id,
3507 rowid,
3508 record_bytes,
3509 commit_ts,
3510 btree_resident,
3511 } => {
3512 let column_count =
3516 crate::types::ImmutableRecordRef::from_bin_record(&record_bytes).column_count();
3517 let row = crate::with_mv_store_allocation_site!(
3518 RowPayload,
3519 Row::new_table_row_in(
3520 RowID::new(table_id, rowid.row_id.clone()),
3521 &record_bytes,
3522 column_count,
3523 alloc,
3524 )?
3525 );
3526 Ok(StreamingResult::UpsertTableRow {
3527 row,
3528 rowid,
3529 commit_ts,
3530 btree_resident,
3531 })
3532 }
3533 ParsedOp::DeleteTable {
3534 rowid,
3535 record_bytes: _,
3536 pk_record_bytes: _,
3537 commit_ts,
3538 btree_resident,
3539 } => Ok(StreamingResult::DeleteTableRow {
3540 rowid,
3541 commit_ts,
3542 btree_resident,
3543 }),
3544 ParsedOp::UpsertIndex {
3545 table_id,
3546 payload,
3547 commit_ts,
3548 btree_resident,
3549 } => {
3550 let key_record = crate::types::ImmutableRecord::from_bin_record(payload);
3551 let column_count = key_record.column_count();
3552 let index_info = get_index_info(table_id, IndexOpKind::Upsert)?;
3553 let key = Arc::new(SortableIndexKey::new_from_record(key_record, index_info));
3554 let rowid = RowID::new(table_id, RowKey::Record(key));
3555 let row = Row::new_index_row(rowid.clone(), column_count);
3556 Ok(StreamingResult::UpsertIndexRow {
3557 row,
3558 rowid,
3559 commit_ts,
3560 btree_resident,
3561 })
3562 }
3563 ParsedOp::DeleteIndex {
3564 table_id,
3565 payload,
3566 commit_ts,
3567 btree_resident,
3568 } => {
3569 let key_record = crate::types::ImmutableRecord::from_bin_record(payload);
3570 let column_count = key_record.column_count();
3571 let index_info = get_index_info(table_id, IndexOpKind::Delete)?;
3572 let key = Arc::new(SortableIndexKey::new_from_record(key_record, index_info));
3573 let rowid = RowID::new(table_id, RowKey::Record(key));
3574 let row = Row::new_index_row(rowid.clone(), column_count);
3575 Ok(StreamingResult::DeleteIndexRow {
3576 row,
3577 rowid,
3578 commit_ts,
3579 btree_resident,
3580 })
3581 }
3582 ParsedOp::UpdateHeader { header, commit_ts } => {
3583 Ok(StreamingResult::UpdateHeader { header, commit_ts })
3584 }
3585 }
3586 }
3587
3588 fn remaining_bytes(&self) -> usize {
3589 let bytes_in_buffer = self.bytes_can_read();
3590 let bytes_in_file = self.file_size.saturating_sub(self.offset);
3591 bytes_in_buffer + bytes_in_file
3592 }
3593
3594 fn try_consume_bytes(&mut self, amount: usize) -> Result<IOResult<Option<Vec<u8>>>> {
3595 if self.remaining_bytes() < amount {
3596 return Ok(IOResult::Done(None));
3597 }
3598 return_if_io!(self.read_more_data(amount));
3599 let buffer = self.buffer.read();
3600 let start = self.buffer_offset;
3601 let end = start + amount;
3602 let bytes = buffer[start..end].to_vec();
3603 self.buffer_offset = end;
3604 Ok(IOResult::Done(Some(bytes)))
3605 }
3606
3607 fn try_consume_fixed<const N: usize>(&mut self) -> Result<IOResult<Option<[u8; N]>>> {
3608 if self.remaining_bytes() < N {
3609 return Ok(IOResult::Done(None));
3610 }
3611 return_if_io!(self.read_more_data(N));
3612 let buffer = self.buffer.read();
3613 let start = self.buffer_offset;
3614 let end = start + N;
3615 let mut out = [0u8; N];
3616 out.copy_from_slice(&buffer[start..end]);
3617 self.buffer_offset = end;
3618 Ok(IOResult::Done(Some(out)))
3619 }
3620
3621 fn try_consume_u8(&mut self) -> Result<IOResult<Option<u8>>> {
3622 if self.remaining_bytes() == 0 {
3623 return Ok(IOResult::Done(None));
3624 }
3625 return_if_io!(self.read_more_data(1));
3626 let r = self.buffer.read()[self.buffer_offset];
3627 self.buffer_offset += 1;
3628 Ok(IOResult::Done(Some(r)))
3629 }
3630
3631 #[allow(clippy::type_complexity)]
3638 fn consume_varint_bytes(&mut self) -> Result<IOResult<Option<(u64, [u8; 9], usize)>>> {
3639 let mut v: u64 = 0;
3640 let mut bytes = [0u8; 9];
3641 let mut len = 0usize;
3642 for _ in 0..8 {
3643 let Some(c) = return_if_io!(self.try_consume_u8()) else {
3644 return Ok(IOResult::Done(None));
3645 };
3646 bytes[len] = c;
3647 len += 1;
3648 v = (v << 7) + (c & 0x7f) as u64;
3649 if (c & 0x80) == 0 {
3650 return Ok(IOResult::Done(Some((v, bytes, len))));
3651 }
3652 }
3653 let Some(c) = return_if_io!(self.try_consume_u8()) else {
3654 return Ok(IOResult::Done(None));
3655 };
3656 bytes[len] = c;
3657 len += 1;
3658 if (v >> 48) == 0 {
3659 return Err(LimboError::Corrupt("Invalid varint".to_string()));
3660 }
3661 v = (v << 8) + c as u64;
3662 Ok(IOResult::Done(Some((v, bytes, len))))
3663 }
3664
3665 fn read_exact_at(&mut self, pos: u64, len: usize) -> Result<IOResult<Vec<u8>>> {
3673 loop {
3674 if let Some(InFlightRead::Exact { completion, .. }) = &self.in_flight_read {
3675 if !completion.succeeded() {
3676 let c = completion.clone();
3677 io_yield_one!(c);
3678 }
3679 let Some(InFlightRead::Exact {
3680 out, expected_len, ..
3681 }) = self.in_flight_read.take()
3682 else {
3683 unreachable!("in_flight_read variant just matched Exact");
3684 };
3685 let result = out.read().clone();
3686 if result.len() != expected_len {
3687 return Err(LimboError::Corrupt(format!(
3688 "Logical log short read: expected {expected_len}, got {}",
3689 result.len()
3690 )));
3691 }
3692 return Ok(IOResult::Done(result));
3693 }
3694
3695 let header_buf = Arc::new(Buffer::new_temporary(len));
3696 let out = Arc::new(RwLock::new(Vec::with_capacity(len)));
3697 let out_clone = out.clone();
3698 let completion: Box<ReadComplete> = Box::new(move |res| {
3699 let out = out_clone.clone();
3700 let mut out = out.write();
3701 let Ok((buf, bytes_read)) = res else {
3702 tracing::error!("couldn't read logical log header err={:?}", res);
3703 return None;
3704 };
3705 if bytes_read > 0 {
3706 out.extend_from_slice(&buf.as_slice()[..bytes_read as usize]);
3707 }
3708 None
3709 });
3710 let c = Completion::new_read(header_buf, completion);
3711 let c = self.file.pread(pos, c)?;
3712 self.in_flight_read = Some(InFlightRead::Exact {
3713 completion: c,
3714 out,
3715 expected_len: len,
3716 });
3717 }
3720 }
3721
3722 fn get_buffer(&self) -> crate::sync::RwLockReadGuard<'_, Vec<u8>> {
3723 self.buffer.read()
3724 }
3725
3726 pub fn read_more_data(&mut self, need: usize) -> Result<IOResult<()>> {
3734 loop {
3735 if let Some(InFlightRead::Chunk { completion, .. }) = &self.in_flight_read {
3738 if !completion.succeeded() {
3739 let c = completion.clone();
3740 io_yield_one!(c);
3741 }
3742 let Some(InFlightRead::Chunk { pre_size, .. }) = self.in_flight_read.take() else {
3743 unreachable!("in_flight_read variant just matched Chunk");
3744 };
3745 let buffer_size_after_read = self.buffer.read().len();
3746 let bytes_read = buffer_size_after_read - pre_size;
3747 if bytes_read == 0 {
3748 return Err(LimboError::Corrupt(format!(
3749 "Expected to read more bytes but read 0 bytes at offset {}",
3750 self.offset
3751 )));
3752 }
3753 self.offset += bytes_read;
3754 }
3755
3756 let buffer_size_before_read = self.buffer.read().len();
3757 turso_assert!(
3758 buffer_size_before_read >= self.buffer_offset,
3759 "buffer_size_before_read < buffer_offset",
3760 { "buffer_size_before_read": buffer_size_before_read, "buffer_offset": self.buffer_offset }
3761 );
3762 let bytes_available_in_buffer = buffer_size_before_read - self.buffer_offset;
3763 let still_need = need.saturating_sub(bytes_available_in_buffer);
3764
3765 if still_need == 0 {
3766 return Ok(IOResult::Done(()));
3773 }
3774
3775 let drain_to = self.frame_anchor.min(self.buffer.read().len());
3787 if drain_to > 0 {
3788 let _ = self.buffer.write().drain(0..drain_to);
3789 self.buffer_offset -= drain_to;
3790 self.frame_anchor -= drain_to;
3791 }
3792
3793 turso_assert!(
3794 self.file_size >= self.offset,
3795 "file_size < offset",
3796 { "file_size": self.file_size, "offset": self.offset }
3797 );
3798 let buffer_size_before_read = self.buffer.read().len();
3802 let to_read = 4096.max(still_need).min(self.file_size - self.offset);
3803
3804 if to_read == 0 {
3805 return Err(LimboError::Corrupt(format!(
3807 "Expected to read {still_need} bytes more but reached end of file at offset {}",
3808 self.offset
3809 )));
3810 }
3811
3812 let header_buf = Arc::new(Buffer::new_temporary(to_read));
3813 let buffer = self.buffer.clone();
3814 let completion: Box<ReadComplete> = Box::new(move |res| match res {
3815 Ok((buf, bytes_read)) => {
3816 let mut buffer = buffer.write();
3817 let buf = buf.as_slice();
3818 if bytes_read > 0 {
3819 buffer.extend_from_slice(&buf[..bytes_read as usize]);
3820 }
3821 None
3822 }
3823 Err(err) => Some(err),
3824 });
3825 let c = Completion::new_read(header_buf, completion);
3826 let c = self.file.pread(self.offset as u64, c)?;
3827 self.in_flight_read = Some(InFlightRead::Chunk {
3828 completion: c,
3829 pre_size: buffer_size_before_read,
3830 });
3831 }
3834 }
3835
3836 fn bytes_can_read(&self) -> usize {
3837 self.buffer.read().len().saturating_sub(self.buffer_offset)
3838 }
3839}
3840
3841struct EncryptedPayloadReadContext {
3843 payload_size: usize,
3844 op_count: u32,
3845 commit_ts: u64,
3846 salt: u64,
3847 nonce_size: usize,
3848 tag_size: usize,
3849}
3850
3851enum PayloadParseResult {
3859 Ok(Vec<ParsedOp>, u32),
3861 Eof,
3863}
3864
3865enum EncryptedChunkReadResult {
3869 Ok { running_crc: u32 },
3870 Eof,
3871}
3872
3873#[cfg_attr(clt_turso_tests, derive(Debug))]
3874enum ParseResult {
3875 Frame(ParsedFrame),
3877 Eof,
3879 InvalidFrame,
3884}
3885
3886#[cfg_attr(clt_turso_tests, derive(Debug))]
3887pub struct ParsedFrame {
3888 ops: Vec<ParsedOp>,
3889 pub portable_changes: Vec<u8>,
3890 pub extension_record_count: u32,
3891 pub frame_flags: u32,
3892 pub commit_ts: u64,
3893 pub end_offset: usize,
3894}
3895
3896#[cfg_attr(clt_turso_tests, derive(Debug, PartialEq, Eq))]
3897pub(crate) enum ParsedOp {
3898 UpsertTable {
3899 table_id: MVTableId,
3900 rowid: RowID,
3901 record_bytes: Vec<u8>,
3902 commit_ts: u64,
3903 btree_resident: bool,
3904 },
3905 DeleteTable {
3906 rowid: RowID,
3907 record_bytes: Vec<u8>,
3908 pk_record_bytes: Vec<u8>,
3909 commit_ts: u64,
3910 btree_resident: bool,
3911 },
3912 UpsertIndex {
3913 table_id: MVTableId,
3914 payload: Vec<u8>,
3915 commit_ts: u64,
3916 btree_resident: bool,
3917 },
3918 DeleteIndex {
3919 table_id: MVTableId,
3920 payload: Vec<u8>,
3921 commit_ts: u64,
3922 btree_resident: bool,
3923 },
3924 UpdateHeader {
3925 header: DatabaseHeader,
3926 commit_ts: u64,
3927 },
3928}
3929
3930#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
3931pub(crate) enum IndexOpKind {
3932 Upsert,
3933 Delete,
3934}
3935
3936#[cfg(clt_turso_tests)]
3937mod tests {
3938 use crate::types::IOResult;
3939 use crate::util::IOExt as _;
3940 use std::cell::RefCell;
3941 use std::collections::BTreeSet;
3942 use std::sync::Once;
3943
3944 use quickcheck_macros::quickcheck;
3945 use rand::{random_range, rng, Rng};
3946 use rand_chacha::{
3947 rand_core::{RngCore, SeedableRng},
3948 ChaCha8Rng,
3949 };
3950
3951 use crate::io::MemoryIO;
3952 use crate::sync::Arc;
3953 use crate::{
3954 mvcc::database::{
3955 tests::{commit_tx, generate_simple_string_row, MvccTestDbNoConn},
3956 MVTableId, Row, RowID, RowKey, SortableIndexKey,
3957 },
3958 schema::Table,
3959 storage::sqlite3_ondisk::{
3960 read_varint, read_varint_partial, varint_len, write_varint, DatabaseHeader,
3961 },
3962 types::{ImmutableRecord, ImmutableRecordRef, IndexInfo, Text},
3963 Buffer, Completion, SharedBufferData, Value, ValueRef,
3964 };
3965
3966 use super::{
3967 build_encrypted_chunk_aad, encrypted_chunk_blob_size, encrypted_chunk_plaintext_len,
3968 encrypted_payload_blob_size, encrypted_payload_chunk_count, serialize_header_entry,
3969 serialize_op_entry, HeaderReadResult, LogHeader, LogicalLog, ParseResult, ParsedOp,
3970 StreamingLogicalLogReader, ENCRYPTED_CHUNK_AAD_SIZE, ENCRYPTED_PAYLOAD_CHUNK_SIZE,
3971 END_MAGIC, EXT_FRAME_MAGIC, FRAME_MAGIC, LOG_HDR_CRC_START, LOG_HDR_RESERVED_START,
3972 LOG_HDR_SIZE, LOG_VERSION, LOG_VERSION_V2, TX_EXT_HEADER_SIZE, TX_HEADER_SIZE,
3973 TX_HEADER_SIZE_V2, TX_TRAILER_SIZE,
3974 };
3975 #[cfg(clt_turso_feature = "conn_raw_api")]
3976 use super::{EXTENSION_RECORD_HEADER_SIZE, EXTENSION_TYPE_PORTABLE_CHANGES, OP_UPSERT_TABLE};
3977 use crate::OpenFlags;
3978 use crate::{turso_assert, turso_assert_less_than};
3979 use tracing_subscriber::EnvFilter;
3980
3981 fn init_tracing() {
3982 static INIT: Once = Once::new();
3983 INIT.call_once(|| {
3984 let _ = tracing_subscriber::fmt()
3985 .with_env_filter(EnvFilter::from_default_env())
3986 .try_init();
3987 });
3988 }
3989
3990 fn write_single_table_tx(
3991 io: &Arc<dyn crate::IO>,
3992 file_name: &str,
3993 commit_ts: u64,
3994 ) -> (Arc<dyn crate::File>, usize) {
3995 let file = io.open_file(file_name, OpenFlags::Create, false).unwrap();
3996 let mut log = LogicalLog::new(file.clone(), io.clone(), None);
3997
3998 let mut tx = crate::mvcc::database::LogRecord::new(commit_ts);
3999 let row = generate_simple_string_row((-2).into(), 1, "foo");
4000 let version = crate::mvcc::database::RowVersion {
4001 id: 1,
4002 begin: crate::mvcc::database::PackedTs::pack(Some(
4003 crate::mvcc::database::TxTimestampOrID::Timestamp(commit_ts),
4004 )),
4005 end: crate::mvcc::database::PackedTs::pack(None),
4006 row: row.clone(),
4007 btree_resident: false,
4008 materialized_at: crate::mvcc::database::WalPos::ORIGIN,
4009 };
4010 tx.push_row_version_for_test(&version);
4011 let c = log.log_tx(tx).unwrap();
4012 io.wait_for_completion(c).unwrap();
4013
4014 let rowid_len = varint_len(1);
4015 let payload_len = rowid_len + row.payload().len();
4016 let payload_len_len = varint_len(payload_len as u64);
4017 let op_size = 6 + payload_len_len + payload_len;
4018 (file, op_size)
4019 }
4020
4021 #[derive(Debug, Clone, PartialEq, Eq)]
4022 enum ExpectedTableOp {
4023 Upsert {
4024 rowid: i64,
4025 payload: Vec<u8>,
4026 commit_ts: u64,
4027 btree_resident: bool,
4028 },
4029 Delete {
4030 rowid: i64,
4031 commit_ts: u64,
4032 btree_resident: bool,
4033 },
4034 }
4035
4036 fn read_table_ops(file: Arc<dyn crate::File>, io: &Arc<dyn crate::IO>) -> Vec<ExpectedTableOp> {
4037 let mut reader = StreamingLogicalLogReader::new(file, None);
4038 reader.read_header(io).unwrap();
4039 let mut ops = Vec::new();
4040 while let Some(frame) = reader.next_frame_blocking(io).unwrap() {
4041 for op in frame {
4042 match op {
4043 ParsedOp::UpsertTable {
4044 rowid,
4045 record_bytes,
4046 commit_ts,
4047 btree_resident,
4048 ..
4049 } => {
4050 ops.push(ExpectedTableOp::Upsert {
4051 rowid: rowid.row_id.to_int_or_panic(),
4052 payload: record_bytes,
4053 commit_ts,
4054 btree_resident,
4055 });
4056 }
4057 ParsedOp::DeleteTable {
4058 rowid,
4059 commit_ts,
4060 btree_resident,
4061 ..
4062 } => {
4063 ops.push(ExpectedTableOp::Delete {
4064 rowid: rowid.row_id.to_int_or_panic(),
4065 commit_ts,
4066 btree_resident,
4067 });
4068 }
4069 other => panic!("unexpected op: {other:?}"),
4070 }
4071 }
4072 }
4073 ops
4074 }
4075
4076 fn read_file_range_bytes(
4077 file: &Arc<dyn crate::File>,
4078 io: &Arc<dyn crate::IO>,
4079 pos: u64,
4080 len: usize,
4081 ) -> Vec<u8> {
4082 let buf = Arc::new(Buffer::new_temporary(len));
4083 let c = file
4084 .pread(pos, Completion::new_read(buf.clone(), |_| None))
4085 .unwrap();
4086 io.wait_for_completion(c).unwrap();
4087 buf.as_slice().to_vec()
4088 }
4089
4090 #[allow(clippy::too_many_arguments)]
4091 fn append_single_table_op_tx(
4092 log: &mut LogicalLog,
4093 io: &Arc<dyn crate::IO>,
4094 table_id: crate::mvcc::database::MVTableId,
4095 rowid: i64,
4096 commit_ts: u64,
4097 is_delete: bool,
4098 btree_resident: bool,
4099 payload_text: &str,
4100 ) {
4101 let row = generate_simple_string_row(table_id, rowid, payload_text);
4102 let row_version = crate::mvcc::database::RowVersion {
4103 id: commit_ts,
4104 begin: crate::mvcc::database::PackedTs::pack(Some(
4105 crate::mvcc::database::TxTimestampOrID::Timestamp(commit_ts),
4106 )),
4107 end: crate::mvcc::database::PackedTs::pack(if is_delete {
4108 Some(crate::mvcc::database::TxTimestampOrID::Timestamp(commit_ts))
4109 } else {
4110 None
4111 }),
4112 row,
4113 btree_resident,
4114 materialized_at: crate::mvcc::database::WalPos::ORIGIN,
4115 };
4116 let tx = crate::mvcc::database::LogRecord::for_test(commit_ts, &[row_version], None);
4117 let c = log.log_tx(tx).unwrap();
4118 io.wait_for_completion(c).unwrap();
4119 }
4120
4121 fn decode_streaming_varint(bytes: &[u8]) -> crate::Result<Option<(u64, [u8; 9], usize)>> {
4122 let io: Arc<dyn crate::IO> = Arc::new(MemoryIO::new());
4123 let file = io
4124 .open_file("logical_log_varint_decode_tmp", OpenFlags::Create, false)
4125 .unwrap();
4126 let mut reader = StreamingLogicalLogReader::new(file, None);
4127 reader.buffer.write().extend_from_slice(bytes);
4128 io.block(|| reader.consume_varint_bytes())
4129 }
4130
4131 struct SlowReadFile {
4139 data: Vec<u8>,
4140 max_read: usize,
4141 pending: std::sync::Mutex<std::collections::VecDeque<(u64, Completion)>>,
4142 }
4143
4144 impl SlowReadFile {
4145 fn new(data: Vec<u8>, max_read: usize) -> Self {
4146 Self {
4147 data,
4148 max_read,
4149 pending: std::sync::Mutex::new(std::collections::VecDeque::new()),
4150 }
4151 }
4152
4153 fn step(&self) -> bool {
4156 let Some((pos, c)) = self.pending.lock().unwrap().pop_front() else {
4157 return false;
4158 };
4159 let pos = pos as usize;
4160 let read = c.as_read();
4161 let cap = read.buf().len();
4162 let avail = self.data.len().saturating_sub(pos);
4163 let n = cap.min(self.max_read).min(avail);
4164 if n > 0 {
4165 read.buf().as_mut_slice()[..n].copy_from_slice(&self.data[pos..pos + n]);
4166 }
4167 c.complete(n as i32);
4168 true
4169 }
4170 }
4171
4172 impl crate::File for SlowReadFile {
4173 fn lock_file(&self, _exclusive: bool) -> crate::Result<()> {
4174 Ok(())
4175 }
4176 fn unlock_file(&self) -> crate::Result<()> {
4177 Ok(())
4178 }
4179 fn pread(&self, pos: u64, c: Completion) -> crate::Result<Completion> {
4180 self.pending.lock().unwrap().push_back((pos, c.clone()));
4181 Ok(c)
4182 }
4183 fn pwrite(
4184 &self,
4185 _pos: u64,
4186 _buffer: Arc<Buffer>,
4187 _c: Completion,
4188 ) -> crate::Result<Completion> {
4189 unimplemented!("SlowReadFile is read-only")
4190 }
4191 fn sync(
4192 &self,
4193 _c: Completion,
4194 _sync_type: crate::io::FileSyncType,
4195 ) -> crate::Result<Completion> {
4196 unimplemented!("SlowReadFile is read-only")
4197 }
4198 fn size(&self) -> crate::Result<u64> {
4199 Ok(self.data.len() as u64)
4200 }
4201 fn truncate(&self, _len: u64, _c: Completion) -> crate::Result<Completion> {
4202 unimplemented!("SlowReadFile is read-only")
4203 }
4204 }
4205
4206 fn read_file_to_vec(file: &Arc<dyn crate::File>) -> Vec<u8> {
4208 let size = file.size().unwrap() as usize;
4209 let out = Arc::new(std::sync::Mutex::new(Vec::new()));
4210 let sink = out.clone();
4211 let buf = Arc::new(Buffer::new_temporary(size));
4212 let c = Completion::new_read(buf, move |res| {
4213 if let Ok((b, n)) = res {
4214 sink.lock()
4215 .unwrap()
4216 .extend_from_slice(&b.as_slice()[..n as usize]);
4217 }
4218 None
4219 });
4220 let _completion = file.pread(0, c).unwrap();
4221 let bytes = out.lock().unwrap().clone();
4222 assert_eq!(bytes.len(), size, "expected a synchronous full read");
4223 bytes
4224 }
4225
4226 fn table_row_version(
4227 table_id: MVTableId,
4228 rowid: i64,
4229 commit_ts: u64,
4230 data: &str,
4231 ) -> crate::mvcc::database::RowVersion {
4232 crate::mvcc::database::RowVersion {
4233 id: commit_ts,
4234 begin: crate::mvcc::database::PackedTs::pack(Some(
4235 crate::mvcc::database::TxTimestampOrID::Timestamp(commit_ts),
4236 )),
4237 end: crate::mvcc::database::PackedTs::pack(None),
4238 row: generate_simple_string_row(table_id, rowid, data),
4239 btree_resident: false,
4240 materialized_at: crate::mvcc::database::WalPos::ORIGIN,
4241 }
4242 }
4243
4244 fn recover_with_forced_yields(data: Vec<u8>, max_read: usize) -> (Vec<ParsedOp>, usize) {
4248 recover_with_forced_yields_inner(data, max_read, None)
4249 }
4250
4251 fn recover_with_forced_yields_inner(
4252 data: Vec<u8>,
4253 max_read: usize,
4254 encryption: Option<(crate::storage::encryption::EncryptionContext, usize)>,
4255 ) -> (Vec<ParsedOp>, usize) {
4256 let slow = Arc::new(SlowReadFile::new(data, max_read));
4257 let file: Arc<dyn crate::File> = slow.clone();
4258 let mut reader = match encryption {
4259 Some((ctx, chunk_size)) => {
4260 StreamingLogicalLogReader::new_with_payload_chunk_size(file, Some(ctx), chunk_size)
4261 }
4262 None => StreamingLogicalLogReader::new(file, None),
4263 };
4264 let mut peak = 0usize;
4265
4266 loop {
4268 match reader.try_read_header_nonblock().unwrap() {
4269 IOResult::Done(HeaderReadResult::Valid(_)) => break,
4270 IOResult::Done(other) => panic!("unexpected header result: {other:?}"),
4271 IOResult::IO(_) => assert!(slow.step(), "stalled reading header"),
4272 }
4273 peak = peak.max(reader.buffer.read().len());
4274 }
4275
4276 let mut ops = Vec::new();
4278 loop {
4279 let result = reader.next_frame().unwrap();
4280 peak = peak.max(reader.buffer.read().len());
4281 match result {
4282 IOResult::Done(Some(frame_ops)) => ops.extend(frame_ops),
4283 IOResult::Done(None) => break,
4284 IOResult::IO(_) => {
4285 assert!(slow.step(), "stalled reading frame");
4286 peak = peak.max(reader.buffer.read().len());
4287 }
4288 }
4289 }
4290 (ops, peak)
4291 }
4292
4293 #[test]
4301 fn test_logical_log_streaming_recovery_forced_yields_bounded_memory() {
4302 init_tracing();
4303 let io: Arc<dyn crate::IO> = Arc::new(MemoryIO::new());
4304 let file = io
4305 .open_file("logical_log_forced_yields", OpenFlags::Create, false)
4306 .unwrap();
4307 let mut log = LogicalLog::new(file.clone(), io.clone(), None);
4308
4309 let table_id: MVTableId = (-100).into();
4310
4311 let big_payload = "x".repeat(3000);
4315 let frames: Vec<Vec<crate::mvcc::database::RowVersion>> = vec![
4316 vec![table_row_version(table_id, 1, 10, "first")],
4317 (0..40)
4318 .map(|i| table_row_version(table_id, 100 + i as i64, 20, &big_payload))
4319 .collect(),
4320 vec![
4321 table_row_version(table_id, 2, 30, "a"),
4322 table_row_version(table_id, 3, 30, "b"),
4323 ],
4324 ];
4325 for (idx, rows) in frames.iter().enumerate() {
4326 let commit_ts = (idx as u64 + 1) * 10;
4327 let tx = crate::mvcc::database::LogRecord::for_test(commit_ts, rows, None);
4328 let c = log.log_tx(tx).unwrap();
4329 io.wait_for_completion(c).unwrap();
4330 }
4331
4332 let mut expected_reader = StreamingLogicalLogReader::new(file.clone(), None);
4334 expected_reader.read_header(&io).unwrap();
4335 let mut expected = Vec::new();
4336 while let Some(frame) = expected_reader.next_frame_blocking(&io).unwrap() {
4337 expected.extend(frame);
4338 }
4339 assert_eq!(expected.len(), 1 + 40 + 2);
4340
4341 let bytes = read_file_to_vec(&file);
4344 let (recovered, peak) = recover_with_forced_yields(bytes, 64);
4345
4346 assert_eq!(
4347 recovered, expected,
4348 "forced-yield recovery must match the synchronous path exactly"
4349 );
4350
4351 assert!(
4356 peak < 16 * 1024,
4357 "peak buffer {peak} bytes should be bounded to ~one op, not the whole frame"
4358 );
4359 }
4360
4361 #[test]
4372 fn test_encrypted_log_recovery_forced_yields() {
4373 init_tracing();
4374 let io: Arc<dyn crate::IO> = Arc::new(MemoryIO::new());
4375 let enc_ctx = test_enc_ctx();
4376 const TEST_CHUNK_SIZE: usize = 2 * 1024;
4377
4378 let big = "z".repeat(2500);
4381 let tx0 = crate::mvcc::database::LogRecord::for_test(
4382 100,
4383 &[
4384 make_test_row_version((-2).into(), 1, &big, 100),
4385 make_test_row_version((-2).into(), 2, &big, 100),
4386 make_test_row_version((-2).into(), 3, "small", 100),
4387 ],
4388 None,
4389 );
4390 let tx1 = crate::mvcc::database::LogRecord::for_test(
4391 200,
4392 &[make_test_row_version((-2).into(), 4, "tail", 200)],
4393 None,
4394 );
4395 let file = write_encrypted_txs_with_chunk_size_for_test(
4396 &io,
4397 "enc-forced-yields.db-log",
4398 &enc_ctx,
4399 TEST_CHUNK_SIZE,
4400 vec![tx0, tx1],
4401 );
4402
4403 let expected: Vec<ParsedOp> = parse_all_encrypted_tx_ops_with_chunk_size_for_test(
4404 file.clone(),
4405 &io,
4406 &enc_ctx,
4407 TEST_CHUNK_SIZE,
4408 )
4409 .unwrap()
4410 .into_iter()
4411 .flatten()
4412 .collect();
4413 assert_eq!(expected.len(), 4);
4414
4415 let bytes = read_file_to_vec(&file);
4416 let (recovered, _peak) =
4417 recover_with_forced_yields_inner(bytes, 64, Some((enc_ctx.clone(), TEST_CHUNK_SIZE)));
4418
4419 assert_eq!(
4420 recovered, expected,
4421 "encrypted forced-yield recovery must match the synchronous path exactly"
4422 );
4423 }
4424
4425 #[test]
4428 fn test_logical_log_read() {
4429 init_tracing();
4430 let mut db = MvccTestDbNoConn::new_with_random_db();
4433 {
4434 let conn = db.connect();
4435 let pager = conn.pager.load().clone();
4436 let mvcc_store = db.get_mvcc_store();
4437 let table_id: MVTableId = (-100).into();
4438 let tx_id = mvcc_store.begin_tx(pager).unwrap();
4439 let data = ImmutableRecord::from_values(
4441 &[
4442 Value::Text(Text::new("table")), Value::Text(Text::new("test")), Value::Text(Text::new("test")), Value::from_i64(table_id.into()), Value::Text(Text::new(
4447 "CREATE TABLE test(id INTEGER PRIMARY KEY, data TEXT)",
4448 )), ],
4450 5,
4451 )
4452 .unwrap();
4453 mvcc_store
4454 .insert(
4455 tx_id,
4456 Row::new_table_row(
4457 RowID::new((-1).into(), RowKey::Int(1000)),
4458 data.as_blob(),
4459 5,
4460 )
4461 .unwrap(),
4462 )
4463 .unwrap();
4464 let row = generate_simple_string_row(table_id, 1, "foo");
4466 mvcc_store.insert(tx_id, row).unwrap();
4467 commit_tx(mvcc_store, &conn, tx_id).unwrap();
4468 }
4469
4470 db.restart();
4472
4473 let conn = db.connect();
4475 let pager = conn.pager.load().clone();
4476 let mvcc_store = db.get_mvcc_store();
4477 let tx = mvcc_store.begin_tx(pager).unwrap();
4478 let row = mvcc_store
4479 .read(tx, &RowID::new((-100).into(), RowKey::Int(1)))
4480 .unwrap()
4481 .unwrap();
4482 let record = ImmutableRecordRef::from_bin_record(row.payload());
4483 let foo = record.iter().unwrap().next().unwrap().unwrap();
4484 let ValueRef::Text(foo) = foo else {
4485 unreachable!()
4486 };
4487 assert_eq!(foo.as_str(), "foo");
4488 }
4489
4490 #[test]
4493 fn test_logical_log_read_multiple_transactions() {
4494 init_tracing();
4495 let table_id: MVTableId = (-100).into();
4496 let values = (0..100)
4497 .map(|i| {
4498 (
4499 RowID::new(table_id, RowKey::Int(i as i64)),
4500 format!("foo_{i}"),
4501 )
4502 })
4503 .collect::<Vec<(RowID, String)>>();
4504 let mut db = MvccTestDbNoConn::new_with_random_db();
4506 {
4507 let conn = db.connect();
4508 let pager = conn.pager.load().clone();
4509 let mvcc_store = db.get_mvcc_store();
4510
4511 let tx_id = mvcc_store.begin_tx(pager.clone()).unwrap();
4512 let data = ImmutableRecord::from_values(
4514 &[
4515 Value::Text(Text::new("table")), Value::Text(Text::new("test")), Value::Text(Text::new("test")), Value::from_i64(table_id.into()), Value::Text(Text::new(
4520 "CREATE TABLE test(id INTEGER PRIMARY KEY, data TEXT)",
4521 )), ],
4523 5,
4524 )
4525 .unwrap();
4526 mvcc_store
4527 .insert(
4528 tx_id,
4529 Row::new_table_row(
4530 RowID::new((-1).into(), RowKey::Int(1000)),
4531 data.as_blob(),
4532 5,
4533 )
4534 .unwrap(),
4535 )
4536 .unwrap();
4537 commit_tx(mvcc_store.clone(), &conn, tx_id).unwrap();
4538 for (rowid, value) in &values {
4541 let tx_id = mvcc_store.begin_tx(pager.clone()).unwrap();
4542 let row = generate_simple_string_row(
4543 rowid.table_id,
4544 rowid.row_id.to_int_or_panic(),
4545 value,
4546 );
4547 mvcc_store.insert(tx_id, row).unwrap();
4548 commit_tx(mvcc_store.clone(), &conn, tx_id).unwrap();
4549 }
4550 }
4551
4552 db.restart();
4554
4555 let conn = db.connect();
4557 let pager = conn.pager.load().clone();
4558 let mvcc_store = db.get_mvcc_store();
4559 for (rowid, value) in &values {
4560 let tx = mvcc_store.begin_tx(pager.clone()).unwrap();
4561 let row = mvcc_store.read(tx, rowid).unwrap().unwrap();
4562 let record = ImmutableRecordRef::from_bin_record(row.payload());
4563 let foo = record.iter().unwrap().next().unwrap().unwrap();
4564 let ValueRef::Text(foo) = foo else {
4565 unreachable!()
4566 };
4567 assert_eq!(foo.as_str(), value.as_str());
4568 }
4569 }
4570
4571 #[test]
4574 fn test_logical_log_read_fuzz() {
4575 init_tracing();
4576 let table_id: MVTableId = (-100).into();
4577 let seed = rng().random();
4578 let mut rng = ChaCha8Rng::seed_from_u64(seed);
4579 let num_transactions = rng.next_u64() % 128;
4580 let mut txns = vec![];
4581 let mut present_rowids = BTreeSet::new();
4582 let mut non_present_rowids = BTreeSet::new();
4583 for _ in 0..num_transactions {
4584 let num_operations = rng.next_u64() % 8;
4585 let mut ops = vec![];
4586 for _ in 0..num_operations {
4587 let op_type = rng.next_u64() % 2;
4588 match op_type {
4589 0 => {
4590 let row_id = (rng.next_u64() % (i64::MAX as u64)) as i64;
4592 let rowid = RowID::new(table_id, RowKey::Int(row_id));
4593 let row = generate_simple_string_row(
4594 rowid.table_id,
4595 rowid.row_id.to_int_or_panic(),
4596 &format!("row_{row_id}"),
4597 );
4598 ops.push((true, Some(row), rowid.clone()));
4599 present_rowids.insert(rowid.clone());
4600 non_present_rowids.remove(&rowid);
4601 tracing::debug!("insert {rowid:?}");
4602 }
4603 1 => {
4604 if present_rowids.is_empty() {
4605 continue;
4606 }
4607 let row_id_pos = rng.next_u64() as usize % present_rowids.len();
4608 let row_id = present_rowids.iter().nth(row_id_pos).unwrap().clone();
4609 ops.push((false, None, row_id.clone()));
4610 present_rowids.remove(&row_id);
4611 non_present_rowids.insert(row_id.clone());
4612 tracing::debug!("removed {row_id:?}");
4613 }
4614 _ => unreachable!(),
4615 }
4616 }
4617 txns.push(ops);
4618 }
4619 let mut db = MvccTestDbNoConn::new_with_random_db();
4621 let pager = {
4622 let conn = db.connect();
4623 let pager = conn.pager.load().clone();
4624 let mvcc_store = db.get_mvcc_store();
4625
4626 let tx_id = mvcc_store.begin_tx(pager.clone()).unwrap();
4628 let data = ImmutableRecord::from_values(
4629 &[
4630 Value::Text(Text::new("table")), Value::Text(Text::new("test")), Value::Text(Text::new("test")), Value::from_i64(table_id.into()), Value::Text(Text::new(
4635 "CREATE TABLE test(id INTEGER PRIMARY KEY, data TEXT)",
4636 )), ],
4638 5,
4639 )
4640 .unwrap();
4641 mvcc_store
4642 .insert(
4643 tx_id,
4644 Row::new_table_row(
4645 RowID::new((-1).into(), RowKey::Int(1000)),
4646 data.as_blob(),
4647 5,
4648 )
4649 .unwrap(),
4650 )
4651 .unwrap();
4652 commit_tx(mvcc_store.clone(), &conn, tx_id).unwrap();
4653
4654 for ops in &txns {
4656 let tx_id = mvcc_store.begin_tx(pager.clone()).unwrap();
4657 for (is_insert, maybe_row, rowid) in ops {
4658 if *is_insert {
4659 mvcc_store
4660 .insert(tx_id, maybe_row.as_ref().unwrap().clone())
4661 .unwrap();
4662 } else {
4663 mvcc_store.delete(tx_id, rowid.clone()).unwrap();
4664 }
4665 }
4666 commit_tx(mvcc_store.clone(), &conn, tx_id).unwrap();
4667 }
4668
4669 conn.close().unwrap();
4670 pager
4671 };
4672
4673 db.restart();
4674
4675 let _conn = db.connect();
4677 let mvcc_store = db.get_mvcc_store();
4678
4679 let tx = mvcc_store.begin_tx(pager.clone()).unwrap();
4681 for present_rowid in present_rowids {
4682 let row = mvcc_store.read(tx, &present_rowid).unwrap().unwrap();
4683 let record = ImmutableRecordRef::from_bin_record(row.payload());
4684 let foo = record.iter().unwrap().next().unwrap().unwrap();
4685 let ValueRef::Text(foo) = foo else {
4686 unreachable!()
4687 };
4688
4689 assert_eq!(
4690 foo.as_str(),
4691 format!("row_{}", present_rowid.row_id.to_int_or_panic())
4692 );
4693 }
4694
4695 let tx = mvcc_store.begin_tx(pager).unwrap();
4697 for present_rowid in non_present_rowids {
4698 let row = mvcc_store.read(tx, &present_rowid).unwrap();
4699 assert!(
4700 row.is_none(),
4701 "row {present_rowid:?} should have been removed"
4702 );
4703 }
4704 }
4705
4706 #[test]
4709 fn test_logical_log_read_table_and_index_rows() {
4710 init_tracing();
4711 let mut db = MvccTestDbNoConn::new_with_random_db();
4713 {
4714 let conn = db.connect();
4715
4716 conn.execute("CREATE TABLE test(id INTEGER PRIMARY KEY, data TEXT)")
4718 .unwrap();
4719 conn.execute("CREATE INDEX idx_data ON test(data)").unwrap();
4720
4721 conn.execute("PRAGMA wal_checkpoint(TRUNCATE)").unwrap();
4723
4724 conn.execute("INSERT INTO test(id, data) VALUES (1, 'foo')")
4727 .unwrap();
4728 conn.execute("INSERT INTO test(id, data) VALUES (2, 'bar')")
4729 .unwrap();
4730 conn.execute("INSERT INTO test(id, data) VALUES (3, 'baz')")
4731 .unwrap();
4732 }
4733
4734 db.restart();
4736
4737 let conn = db.connect();
4739 let pager = conn.pager.load().clone();
4740 let mvcc_store = db.get_mvcc_store();
4741 let schema = conn.schema.read();
4742 let table = schema.get_table("test").expect("table test should exist");
4743 let Table::BTree(table) = table.as_ref() else {
4744 panic!("table test should be btree");
4745 };
4746 let table_id = mvcc_store.get_table_id_from_root_page(table.root_page);
4747
4748 let index = schema
4750 .get_index("test", "idx_data")
4751 .expect("Index should exist");
4752 let index_id = mvcc_store.get_table_id_from_root_page(index.root_page);
4754 let index_info = Arc::new(IndexInfo::new_from_index(index).unwrap());
4755
4756 let tx = mvcc_store.begin_tx(pager).unwrap();
4758 for (row_id, expected_data) in [(1, "foo"), (2, "bar"), (3, "baz")] {
4759 let row = mvcc_store
4760 .read(tx, &RowID::new(table_id, RowKey::Int(row_id)))
4761 .unwrap()
4762 .expect("Table row should exist");
4763 let record = ImmutableRecordRef::from_bin_record(row.payload());
4764 let values = record.get_values().unwrap();
4765 let data_value = values.get(1).expect("Should have data column");
4766 let ValueRef::Text(data_text) = data_value else {
4767 panic!("Data column should be text");
4768 };
4769 assert_eq!(data_text.as_str(), expected_data);
4770 }
4771
4772 for (row_id, data_value) in [(1, "foo"), (2, "bar"), (3, "baz")] {
4776 let key_record = ImmutableRecord::from_values(
4779 &[
4780 Value::Text(Text::new(data_value.to_string())),
4781 Value::from_i64(row_id),
4782 ],
4783 2,
4784 )
4785 .unwrap();
4786 let sortable_key = SortableIndexKey::new_from_record(key_record, index_info.clone());
4787 let index_rowid = RowID::new(index_id, RowKey::Record(Arc::new(sortable_key)));
4788
4789 let index_row_opt = mvcc_store
4792 .read_from_table_or_index(tx, &index_rowid, Some(index_id))
4793 .unwrap_or_else(|e| {
4794 panic!("Failed to read index row for ({}, {}): {:?}. Index ID: {:?}, root_page: {}",
4795 data_value, row_id, e, index_id, index.root_page)
4796 });
4797
4798 let Some(index_row) = index_row_opt else {
4799 panic!(
4800 "Index row for ({data_value}, {row_id}) not found after recovery. Index rows should be in the logical log."
4801 );
4802 };
4803 let RowKey::Record(sortable_key) = index_row.id.row_id else {
4805 panic!("Index row should have a record row_id");
4806 };
4807 let record = sortable_key.key.clone();
4808 let values = record.get_values().unwrap();
4809 assert_eq!(
4810 values.len(),
4811 2,
4812 "Index row should have 2 columns (data, rowid)"
4813 );
4814 let ValueRef::Text(index_data) = values[0] else {
4815 panic!("First index column should be text");
4816 };
4817 assert_eq!(index_data.as_str(), data_value, "Index data should match");
4818 let ValueRef::Numeric(crate::numeric::Numeric::Integer(index_rowid_val)) = values[1]
4819 else {
4820 panic!("Second index column should be integer (rowid)");
4821 };
4822 assert_eq!(index_rowid_val, row_id, "Index rowid should match");
4823 }
4824 }
4825
4826 #[test]
4829 fn test_logical_log_torn_tail_stops_cleanly() {
4830 init_tracing();
4831 let io: Arc<dyn crate::IO> = Arc::new(MemoryIO::new());
4832 let file = io
4833 .open_file("test.db-log", crate::OpenFlags::Create, false)
4834 .unwrap();
4835 let mut log = LogicalLog::new(file.clone(), io.clone(), None);
4836
4837 let row = generate_simple_string_row((-2).into(), 1, "foo");
4838 let rowid_len = varint_len(1);
4839 let payload_len = rowid_len + row.payload().len();
4840 let payload_len_len = varint_len(payload_len as u64);
4841 let op_size = 6 + payload_len_len + payload_len;
4842 let frame_size = TX_HEADER_SIZE + op_size + TX_TRAILER_SIZE;
4843
4844 let mut tx1 = crate::mvcc::database::LogRecord::new(10);
4845 tx1.push_row_version_for_test(&crate::mvcc::database::RowVersion {
4846 id: 1,
4847 begin: crate::mvcc::database::PackedTs::pack(Some(
4848 crate::mvcc::database::TxTimestampOrID::Timestamp(10),
4849 )),
4850 end: crate::mvcc::database::PackedTs::pack(None),
4851 row: row.clone(),
4852 btree_resident: false,
4853 materialized_at: crate::mvcc::database::WalPos::ORIGIN,
4854 });
4855 let c = log.log_tx(tx1).unwrap();
4856 io.wait_for_completion(c).unwrap();
4857
4858 let mut tx2 = crate::mvcc::database::LogRecord::new(20);
4859 tx2.push_row_version_for_test(&crate::mvcc::database::RowVersion {
4860 id: 2,
4861 begin: crate::mvcc::database::PackedTs::pack(Some(
4862 crate::mvcc::database::TxTimestampOrID::Timestamp(20),
4863 )),
4864 end: crate::mvcc::database::PackedTs::pack(None),
4865 row,
4866 btree_resident: false,
4867 materialized_at: crate::mvcc::database::WalPos::ORIGIN,
4868 });
4869 let c = log.log_tx(tx2).unwrap();
4870 io.wait_for_completion(c).unwrap();
4871
4872 let file_size = file.size().unwrap() as usize;
4873 let last_frame_start = LOG_HDR_SIZE + frame_size;
4874
4875 for cut in (last_frame_start..file_size).rev() {
4877 let c = file
4878 .truncate(cut as u64, Completion::new_trunc(|_| {}))
4879 .unwrap();
4880 io.wait_for_completion(c).unwrap();
4881
4882 let mut reader = StreamingLogicalLogReader::new(file.clone(), None);
4883 reader.read_header(&io).unwrap();
4884 let mut seen = 0;
4885 loop {
4886 match reader.next_frame_blocking(&io) {
4887 Ok(Some(frame)) => {
4888 for op in frame {
4889 match op {
4890 ParsedOp::UpsertTable { .. } => seen += 1,
4891 other => panic!("unexpected op: {other:?}"),
4892 }
4893 }
4894 }
4895 Ok(None) => break,
4896 Err(err) => panic!("unexpected error: {err:?}"),
4897 }
4898 }
4899 assert_eq!(seen, 1, "should apply only the first transaction");
4900 }
4901 }
4902
4903 #[test]
4906 fn test_logical_log_torn_tail_multiple_frames_stops_cleanly() {
4907 init_tracing();
4908 let io: Arc<dyn crate::IO> = Arc::new(MemoryIO::new());
4909 let file = io
4910 .open_file(
4911 "logical_log_torn_tail_multi_frame",
4912 OpenFlags::Create,
4913 false,
4914 )
4915 .unwrap();
4916 let mut log = LogicalLog::new(file.clone(), io.clone(), None);
4917
4918 append_single_table_op_tx(&mut log, &io, (-2).into(), 1, 1, false, false, "a");
4919 append_single_table_op_tx(&mut log, &io, (-2).into(), 2, 2, false, false, "b");
4920 let after_tx2 = log.offset as usize;
4921 append_single_table_op_tx(&mut log, &io, (-2).into(), 3, 3, false, false, "c");
4922 let after_tx3 = log.offset as usize;
4923
4924 let partial_tail_len = (after_tx3 - after_tx2) / 2;
4925 let trunc_offset = (after_tx2 + partial_tail_len) as u64;
4926 let c = file
4927 .truncate(trunc_offset, Completion::new_trunc(|_| {}))
4928 .unwrap();
4929 io.wait_for_completion(c).unwrap();
4930
4931 let read_back = read_table_ops(file.clone(), &io);
4932 assert_eq!(read_back.len(), 2);
4933 assert_eq!(
4934 read_back[0],
4935 ExpectedTableOp::Upsert {
4936 rowid: 1,
4937 payload: generate_simple_string_row((-2).into(), 1, "a")
4938 .payload()
4939 .to_vec(),
4940 commit_ts: 1,
4941 btree_resident: false,
4942 }
4943 );
4944 assert_eq!(
4945 read_back[1],
4946 ExpectedTableOp::Upsert {
4947 rowid: 2,
4948 payload: generate_simple_string_row((-2).into(), 2, "b")
4949 .payload()
4950 .to_vec(),
4951 commit_ts: 2,
4952 btree_resident: false,
4953 }
4954 );
4955 }
4956
4957 #[test]
4960 fn test_logical_log_read_i32_min_table_id() {
4961 init_tracing();
4962 let io: Arc<dyn crate::IO> = Arc::new(MemoryIO::new());
4963 let file = io
4964 .open_file("logical_log_i32_min_table_id", OpenFlags::Create, false)
4965 .unwrap();
4966 let mut log = LogicalLog::new(file.clone(), io.clone(), None);
4967 let table_id = crate::mvcc::database::MVTableId::from(i32::MIN as i64);
4968
4969 append_single_table_op_tx(&mut log, &io, table_id, 7, 11, false, false, "min");
4970
4971 let mut reader = StreamingLogicalLogReader::new(file, None);
4972 reader.read_header(&io).unwrap();
4973 let frame = reader
4974 .next_frame_blocking(&io)
4975 .unwrap()
4976 .expect("expected one frame");
4977 assert_eq!(frame.len(), 1);
4978 match &frame[0] {
4979 ParsedOp::UpsertTable { rowid, .. } => {
4980 assert_eq!(rowid.table_id, table_id);
4981 assert_eq!(rowid.row_id.to_int_or_panic(), 7);
4982 }
4983 other => panic!("unexpected op: {other:?}"),
4984 }
4985 }
4986
4987 #[test]
4995 fn test_logical_log_rowid_negative_varint_roundtrip() {
4996 init_tracing();
4997 let io: Arc<dyn crate::IO> = Arc::new(MemoryIO::new());
4998 let file = io
4999 .open_file(
5000 "logical_log_negative_rowid_roundtrip",
5001 OpenFlags::Create,
5002 false,
5003 )
5004 .unwrap();
5005 let mut log = LogicalLog::new(file.clone(), io.clone(), None);
5006
5007 append_single_table_op_tx(&mut log, &io, (-2).into(), -1, 1, false, false, "neg");
5008 append_single_table_op_tx(&mut log, &io, (-2).into(), -1, 2, true, false, "neg");
5009 let offset_after_frame2 = log.offset;
5010
5011 let row3 = generate_simple_string_row((-2).into(), 3, "deferred");
5013 let tx3 = crate::mvcc::database::LogRecord::for_test(
5014 3,
5015 &[crate::mvcc::database::RowVersion {
5016 id: 3,
5017 begin: crate::mvcc::database::PackedTs::pack(Some(
5018 crate::mvcc::database::TxTimestampOrID::Timestamp(3),
5019 )),
5020 end: crate::mvcc::database::PackedTs::pack(None),
5021 row: row3,
5022 btree_resident: false,
5023 materialized_at: crate::mvcc::database::WalPos::ORIGIN,
5024 }],
5025 None,
5026 );
5027 let (c, bytes_written) = log.log_tx_deferred_offset(tx3, None).unwrap();
5028 io.wait_for_completion(c).unwrap();
5029
5030 assert_eq!(
5031 log.offset, offset_after_frame2,
5032 "deferred write must not advance offset before advance_offset_after_success"
5033 );
5034 log.advance_offset_after_success(bytes_written);
5035 assert_eq!(
5036 log.offset,
5037 offset_after_frame2 + bytes_written,
5038 "offset must advance by exactly bytes_written after confirmation"
5039 );
5040
5041 let read_back = read_table_ops(file, &io);
5042 assert_eq!(read_back.len(), 3);
5043 match &read_back[0] {
5044 ExpectedTableOp::Upsert { rowid, .. } => assert_eq!(*rowid, -1),
5045 other => panic!("unexpected op: {other:?}"),
5046 }
5047 match &read_back[1] {
5048 ExpectedTableOp::Delete { rowid, .. } => assert_eq!(*rowid, -1),
5049 other => panic!("unexpected op: {other:?}"),
5050 }
5051 match &read_back[2] {
5052 ExpectedTableOp::Upsert { rowid, .. } => assert_eq!(*rowid, 3),
5053 other => panic!("unexpected op: {other:?}"),
5054 }
5055 }
5056
5057 #[test]
5058 fn test_on_serialization_complete_gets_shared_write_bytes() {
5059 init_tracing();
5060 let io: Arc<dyn crate::IO> = Arc::new(MemoryIO::new());
5061 let file = io
5062 .open_file(
5063 "serialization-callback-shared.db-log",
5064 OpenFlags::Create,
5065 false,
5066 )
5067 .unwrap();
5068 let mut log = LogicalLog::new(file.clone(), io.clone(), None);
5069 let captured = RefCell::new(Vec::<(SharedBufferData, u32)>::new());
5070 let callback = |bytes: SharedBufferData, crc: u32| {
5071 captured.borrow_mut().push((bytes, crc));
5072 Ok(())
5073 };
5074
5075 let tx1 = crate::mvcc::database::LogRecord::for_test(
5076 1,
5077 &[crate::mvcc::database::RowVersion {
5078 id: 1,
5079 begin: crate::mvcc::database::PackedTs::pack(Some(
5080 crate::mvcc::database::TxTimestampOrID::Timestamp(1),
5081 )),
5082 end: crate::mvcc::database::PackedTs::pack(None),
5083 row: generate_simple_string_row((-2).into(), 1, "first"),
5084 btree_resident: false,
5085 materialized_at: crate::mvcc::database::WalPos::ORIGIN,
5086 }],
5087 None,
5088 );
5089 let (c, first_len) = log.log_tx_deferred_offset(tx1, Some(&callback)).unwrap();
5090 io.wait_for_completion(c).unwrap();
5091 log.advance_offset_after_success(first_len);
5092
5093 let tx2 = crate::mvcc::database::LogRecord::for_test(
5094 2,
5095 &[crate::mvcc::database::RowVersion {
5096 id: 2,
5097 begin: crate::mvcc::database::PackedTs::pack(Some(
5098 crate::mvcc::database::TxTimestampOrID::Timestamp(2),
5099 )),
5100 end: crate::mvcc::database::PackedTs::pack(None),
5101 row: generate_simple_string_row((-2).into(), 2, "second"),
5102 btree_resident: false,
5103 materialized_at: crate::mvcc::database::WalPos::ORIGIN,
5104 }],
5105 None,
5106 );
5107 let (c, second_len) = log.log_tx_deferred_offset(tx2, Some(&callback)).unwrap();
5108 io.wait_for_completion(c).unwrap();
5109 log.advance_offset_after_success(second_len);
5110
5111 let captured = captured.borrow();
5112 assert_eq!(captured.len(), 2);
5113 assert_eq!(captured[0].0.len(), first_len as usize);
5114 assert_eq!(captured[1].0.len(), second_len as usize);
5115 assert!(matches!(&captured[0].0, SharedBufferData::Full(_)));
5116 assert!(matches!(&captured[1].0, SharedBufferData::View(_)));
5117
5118 let first_on_disk = read_file_range_bytes(&file, &io, 0, first_len as usize);
5119 let second_on_disk = read_file_range_bytes(&file, &io, first_len, second_len as usize);
5120 assert_eq!(captured[0].0.as_slice(), first_on_disk.as_slice());
5121 assert_eq!(captured[1].0.as_slice(), second_on_disk.as_slice());
5122 assert_eq!(
5123 captured[0].1,
5124 u32::from_le_bytes(
5125 captured[0].0.as_slice()[captured[0].0.len() - TX_TRAILER_SIZE
5126 ..captured[0].0.len() - TX_TRAILER_SIZE + 4]
5127 .try_into()
5128 .unwrap()
5129 )
5130 );
5131 assert_eq!(
5132 captured[1].1,
5133 u32::from_le_bytes(
5134 captured[1].0.as_slice()[captured[1].0.len() - TX_TRAILER_SIZE
5135 ..captured[1].0.len() - TX_TRAILER_SIZE + 4]
5136 .try_into()
5137 .unwrap()
5138 )
5139 );
5140 }
5141
5142 #[test]
5145 fn test_logical_log_corruption_detected() {
5146 init_tracing();
5147 let io: Arc<dyn crate::IO> = Arc::new(MemoryIO::new());
5148 let file = io
5149 .open_file("corrupt.db-log", crate::OpenFlags::Create, false)
5150 .unwrap();
5151 let mut log = LogicalLog::new(file.clone(), io.clone(), None);
5152
5153 let mut tx = crate::mvcc::database::LogRecord::new(123);
5154 let row = generate_simple_string_row((-2).into(), 1, "foo");
5155 let version = crate::mvcc::database::RowVersion {
5156 id: 1,
5157 begin: crate::mvcc::database::PackedTs::pack(Some(
5158 crate::mvcc::database::TxTimestampOrID::Timestamp(123),
5159 )),
5160 end: crate::mvcc::database::PackedTs::pack(None),
5161 row,
5162 btree_resident: false,
5163 materialized_at: crate::mvcc::database::WalPos::ORIGIN,
5164 };
5165 tx.push_row_version_for_test(&version);
5166 let c = log.log_tx(tx).unwrap();
5167 io.wait_for_completion(c).unwrap();
5168
5169 let mut reader = StreamingLogicalLogReader::new(file.clone(), None);
5171 reader.read_header(&io).unwrap();
5172 let offset = reader.offset + TX_HEADER_SIZE + 6; let buf = Arc::new(Buffer::new(vec![0xFF]));
5176 let c = file
5177 .pwrite(offset as u64, buf, Completion::new_write(|_| {}))
5178 .unwrap();
5179 io.wait_for_completion(c).unwrap();
5180
5181 let mut reader = StreamingLogicalLogReader::new(file.clone(), None);
5182 reader.read_header(&io).unwrap();
5183 let res = reader.next_frame_blocking(&io);
5184 assert!(res.unwrap().is_none());
5185 }
5186
5187 #[test]
5190 fn test_logical_log_payload_len_varint_corrupt_tail_keeps_prefix() {
5191 init_tracing();
5192 let io: Arc<dyn crate::IO> = Arc::new(MemoryIO::new());
5193 let file = io
5194 .open_file(
5195 "payload-len-varint-corrupt.db-log",
5196 OpenFlags::Create,
5197 false,
5198 )
5199 .unwrap();
5200 let mut log = LogicalLog::new(file.clone(), io.clone(), None);
5201
5202 append_single_table_op_tx(&mut log, &io, (-2).into(), 1, 1, false, false, "first");
5203 let frame2_start = log.offset;
5204 append_single_table_op_tx(&mut log, &io, (-2).into(), 2, 2, false, false, "second");
5205
5206 let payload_len_offset = frame2_start + (TX_HEADER_SIZE + 6) as u64;
5208 let mut bad_varint = vec![0x80; 8];
5209 bad_varint.push(0x00);
5210 let c = file
5211 .pwrite(
5212 payload_len_offset,
5213 Arc::new(Buffer::new(bad_varint)),
5214 Completion::new_write(|_| {}),
5215 )
5216 .unwrap();
5217 io.wait_for_completion(c).unwrap();
5218
5219 let read_back = read_table_ops(file, &io);
5220 assert_eq!(read_back.len(), 1);
5221 assert_eq!(
5222 read_back[0],
5223 ExpectedTableOp::Upsert {
5224 rowid: 1,
5225 payload: generate_simple_string_row((-2).into(), 1, "first")
5226 .payload()
5227 .to_vec(),
5228 commit_ts: 1,
5229 btree_resident: false,
5230 }
5231 );
5232 }
5233
5234 #[test]
5237 fn test_logical_log_end_magic_corruption() {
5238 init_tracing();
5239 let io: Arc<dyn crate::IO> = Arc::new(MemoryIO::new());
5240 let (file, op_size) = write_single_table_tx(&io, "end-magic.db-log", 100);
5241 let trailer_offset = LOG_HDR_SIZE + TX_HEADER_SIZE + op_size;
5242 let bad = Arc::new(Buffer::new(0u32.to_le_bytes().to_vec()));
5244 let c = file
5245 .pwrite(
5246 (trailer_offset + 4) as u64,
5247 bad,
5248 Completion::new_write(|_| {}),
5249 )
5250 .unwrap();
5251 io.wait_for_completion(c).unwrap();
5252
5253 let mut reader = StreamingLogicalLogReader::new(file.clone(), None);
5254 reader.read_header(&io).unwrap();
5255 let res = reader.next_frame_blocking(&io);
5256 assert!(res.unwrap().is_none());
5257 }
5258
5259 #[test]
5262 fn test_logical_log_payload_size_corruption() {
5263 init_tracing();
5264 let io: Arc<dyn crate::IO> = Arc::new(MemoryIO::new());
5265 let (file, op_size) = write_single_table_tx(&io, "payload-size.db-log", 101);
5266 let bad_payload_size = (op_size as u64 + 1).to_le_bytes().to_vec();
5269 let bad = Arc::new(Buffer::new(bad_payload_size));
5270 let c = file
5271 .pwrite(
5272 (LOG_HDR_SIZE + 4) as u64,
5273 bad,
5274 Completion::new_write(|_| {}),
5275 )
5276 .unwrap();
5277 io.wait_for_completion(c).unwrap();
5278
5279 let mut reader = StreamingLogicalLogReader::new(file.clone(), None);
5280 reader.read_header(&io).unwrap();
5281 let res = reader.next_frame_blocking(&io);
5282 assert!(res.unwrap().is_none());
5283 }
5284
5285 #[test]
5288 fn test_logical_log_frame_magic_corruption() {
5289 init_tracing();
5290 let io: Arc<dyn crate::IO> = Arc::new(MemoryIO::new());
5291 let (file, _) = write_single_table_tx(&io, "frame-magic.db-log", 103);
5292
5293 let bad = Arc::new(Buffer::new(0u32.to_le_bytes().to_vec()));
5296 let c = file
5297 .pwrite(LOG_HDR_SIZE as u64, bad, Completion::new_write(|_| {}))
5298 .unwrap();
5299 io.wait_for_completion(c).unwrap();
5300
5301 let mut reader = StreamingLogicalLogReader::new(file.clone(), None);
5302 reader.read_header(&io).unwrap();
5303 let res = reader.next_frame_blocking(&io);
5304 assert!(res.unwrap().is_none());
5305 }
5306
5307 #[test]
5310 fn test_logical_log_crc_field_corruption() {
5311 init_tracing();
5312 let io: Arc<dyn crate::IO> = Arc::new(MemoryIO::new());
5313 let (file, op_size) = write_single_table_tx(&io, "crc-field.db-log", 104);
5314 let trailer_offset = LOG_HDR_SIZE + TX_HEADER_SIZE + op_size;
5315 let bad = Arc::new(Buffer::new(0u32.to_le_bytes().to_vec()));
5317 let c = file
5318 .pwrite(trailer_offset as u64, bad, Completion::new_write(|_| {}))
5319 .unwrap();
5320 io.wait_for_completion(c).unwrap();
5321
5322 let mut reader = StreamingLogicalLogReader::new(file.clone(), None);
5323 reader.read_header(&io).unwrap();
5324 let res = reader.next_frame_blocking(&io);
5325 assert!(res.unwrap().is_none());
5326 }
5327
5328 #[test]
5331 fn test_logical_log_corrupt_tail_keeps_valid_prefix() {
5332 init_tracing();
5333 let io: Arc<dyn crate::IO> = Arc::new(MemoryIO::new());
5334 let file = io
5335 .open_file(
5336 "corrupt-tail-prefix.db-log",
5337 crate::OpenFlags::Create,
5338 false,
5339 )
5340 .unwrap();
5341 let mut log = LogicalLog::new(file.clone(), io.clone(), None);
5342
5343 append_single_table_op_tx(&mut log, &io, (-2).into(), 1, 10, false, false, "a");
5344 let after_first = log.offset as usize;
5345 append_single_table_op_tx(&mut log, &io, (-2).into(), 2, 20, false, false, "b");
5346 let after_second = log.offset as usize;
5347 let second_frame_len = after_second - after_first;
5348
5349 let second_trailer_crc_offset = after_first + second_frame_len - TX_TRAILER_SIZE;
5351 let c = file
5352 .pwrite(
5353 second_trailer_crc_offset as u64,
5354 Arc::new(Buffer::new(vec![0xDE, 0xAD, 0xBE, 0xEF])),
5355 Completion::new_write(|_| {}),
5356 )
5357 .unwrap();
5358 io.wait_for_completion(c).unwrap();
5359
5360 let ops = read_table_ops(file, &io);
5361 assert_eq!(
5362 ops,
5363 vec![ExpectedTableOp::Upsert {
5364 rowid: 1,
5365 payload: generate_simple_string_row((-2).into(), 1, "a")
5366 .payload()
5367 .to_vec(),
5368 commit_ts: 10,
5369 btree_resident: false,
5370 }]
5371 );
5372 }
5373
5374 #[test]
5377 fn test_logical_log_header_corruption_detected() {
5378 init_tracing();
5379 let io: Arc<dyn crate::IO> = Arc::new(MemoryIO::new());
5380 let file = io
5381 .open_file("header-corrupt.db-log", crate::OpenFlags::Create, false)
5382 .unwrap();
5383 let mut log = LogicalLog::new(file.clone(), io.clone(), None);
5384 let tx = crate::mvcc::database::LogRecord::for_test(77, &[], None);
5385 let c = log.log_tx(tx).unwrap();
5386 io.wait_for_completion(c).unwrap();
5387
5388 let bad = Arc::new(Buffer::new(0u32.to_le_bytes().to_vec()));
5390 let c = file.pwrite(0, bad, Completion::new_write(|_| {})).unwrap();
5391 io.wait_for_completion(c).unwrap();
5392
5393 let mut reader = StreamingLogicalLogReader::new(file.clone(), None);
5394 let res = reader.read_header(&io);
5395 assert!(res.is_err());
5396 }
5397
5398 #[test]
5401 fn test_logical_log_header_flags_rejected() {
5402 init_tracing();
5403 let io: Arc<dyn crate::IO> = Arc::new(MemoryIO::new());
5404 let (file, _) = write_single_table_tx(&io, "header-flags.db-log", 105);
5405
5406 let c = file
5408 .pwrite(
5409 5,
5410 Arc::new(Buffer::new(vec![0b0000_0010])),
5411 Completion::new_write(|_| {}),
5412 )
5413 .unwrap();
5414 io.wait_for_completion(c).unwrap();
5415
5416 let mut reader = StreamingLogicalLogReader::new(file.clone(), None);
5417 let res = reader.read_header(&io);
5418 assert!(res.is_err());
5419 }
5420
5421 #[test]
5425 fn test_logical_log_header_non_default_len_rejected() {
5426 init_tracing();
5427 let io: Arc<dyn crate::IO> = Arc::new(MemoryIO::new());
5428 let (file, _) = write_single_table_tx(&io, "header-len.db-log", 106);
5429
5430 let header_buf = Arc::new(Buffer::new_temporary(LOG_HDR_SIZE));
5431 let c = file
5432 .pread(0, Completion::new_read(header_buf.clone(), |_| None))
5433 .unwrap();
5434 io.wait_for_completion(c).unwrap();
5435 let original_header_bytes = header_buf.as_slice()[..LOG_HDR_SIZE].to_vec();
5436
5437 let mut header_bytes = original_header_bytes.clone();
5439 header_bytes[6..8].copy_from_slice(&(LOG_HDR_SIZE as u16 + 1).to_le_bytes());
5440 header_bytes[LOG_HDR_CRC_START..LOG_HDR_SIZE].fill(0);
5441 let new_crc = crc32c::crc32c(&header_bytes);
5442 header_bytes[LOG_HDR_CRC_START..LOG_HDR_SIZE].copy_from_slice(&new_crc.to_le_bytes());
5443
5444 let c = file
5445 .pwrite(
5446 0,
5447 Arc::new(Buffer::new(header_bytes)),
5448 Completion::new_write(|_| {}),
5449 )
5450 .unwrap();
5451 io.wait_for_completion(c).unwrap();
5452
5453 let mut reader = StreamingLogicalLogReader::new(file.clone(), None);
5454 let res = reader.read_header(&io);
5455 assert!(res.is_err());
5456
5457 let mut header_bytes = original_header_bytes;
5459 header_bytes[4] = 99; header_bytes[LOG_HDR_CRC_START..LOG_HDR_SIZE].fill(0);
5461 let new_crc = crc32c::crc32c(&header_bytes);
5462 header_bytes[LOG_HDR_CRC_START..LOG_HDR_SIZE].copy_from_slice(&new_crc.to_le_bytes());
5463
5464 let c = file
5465 .pwrite(
5466 0,
5467 Arc::new(Buffer::new(header_bytes)),
5468 Completion::new_write(|_| {}),
5469 )
5470 .unwrap();
5471 io.wait_for_completion(c).unwrap();
5472
5473 let mut reader = StreamingLogicalLogReader::new(file, None);
5474 let result = reader.try_read_header(&io).unwrap();
5475 assert!(
5476 matches!(result, HeaderReadResult::Invalid),
5477 "unknown version header must be rejected as Invalid, got {result:?}"
5478 );
5479 }
5480
5481 #[test]
5484 fn test_logical_log_header_reserved_bytes_rejected() {
5485 init_tracing();
5486 let io: Arc<dyn crate::IO> = Arc::new(MemoryIO::new());
5487 let (file, _) = write_single_table_tx(&io, "header-reserved.db-log", 106);
5488
5489 let header_buf = Arc::new(Buffer::new_temporary(LOG_HDR_SIZE));
5491 let c = file
5492 .pread(0, Completion::new_read(header_buf.clone(), |_| None))
5493 .unwrap();
5494 io.wait_for_completion(c).unwrap();
5495 let mut header_bytes = header_buf.as_slice()[..LOG_HDR_SIZE].to_vec();
5496
5497 header_bytes[LOG_HDR_RESERVED_START] = 1;
5499
5500 header_bytes[LOG_HDR_CRC_START..LOG_HDR_SIZE].fill(0);
5502 let new_crc = crc32c::crc32c(&header_bytes);
5503 header_bytes[LOG_HDR_CRC_START..LOG_HDR_SIZE].copy_from_slice(&new_crc.to_le_bytes());
5504
5505 let c = file
5507 .pwrite(
5508 0,
5509 Arc::new(Buffer::new(header_bytes)),
5510 Completion::new_write(|_| {}),
5511 )
5512 .unwrap();
5513 io.wait_for_completion(c).unwrap();
5514
5515 let mut reader = StreamingLogicalLogReader::new(file.clone(), None);
5516 let res = reader.read_header(&io);
5517 assert!(res.is_err());
5518 }
5519
5520 #[test]
5523 fn test_logical_log_op_reserved_flags_rejected() {
5524 init_tracing();
5525 let io: Arc<dyn crate::IO> = Arc::new(MemoryIO::new());
5526 let (file, _) = write_single_table_tx(&io, "op-flags.db-log", 108);
5527
5528 let c = file
5530 .pwrite(
5531 (LOG_HDR_SIZE + TX_HEADER_SIZE + 1) as u64,
5532 Arc::new(Buffer::new(vec![0b0000_0010])),
5533 Completion::new_write(|_| {}),
5534 )
5535 .unwrap();
5536 io.wait_for_completion(c).unwrap();
5537
5538 let mut reader = StreamingLogicalLogReader::new(file.clone(), None);
5539 reader.read_header(&io).unwrap();
5540 let res = reader.next_frame_blocking(&io);
5541 assert!(res.unwrap().is_none());
5542 }
5543
5544 #[test]
5547 fn test_logical_log_non_negative_table_id_rejected() {
5548 init_tracing();
5549 let io: Arc<dyn crate::IO> = Arc::new(MemoryIO::new());
5550 let (file, _) = write_single_table_tx(&io, "table-id-sign.db-log", 109);
5551
5552 let c = file
5554 .pwrite(
5555 (LOG_HDR_SIZE + TX_HEADER_SIZE + 2) as u64,
5556 Arc::new(Buffer::new(1i32.to_le_bytes().to_vec())),
5557 Completion::new_write(|_| {}),
5558 )
5559 .unwrap();
5560 io.wait_for_completion(c).unwrap();
5561
5562 let mut reader = StreamingLogicalLogReader::new(file.clone(), None);
5563 reader.read_header(&io).unwrap();
5564 let res = reader.next_frame_blocking(&io);
5565 assert!(res.unwrap().is_none());
5566 }
5567
5568 #[test]
5574 fn test_logical_log_empty_transaction_frame() {
5575 init_tracing();
5576 let io: Arc<dyn crate::IO> = Arc::new(MemoryIO::new());
5577 let file = io
5578 .open_file("empty-tx.db-log", crate::OpenFlags::Create, false)
5579 .unwrap();
5580 let mut log = LogicalLog::new(file.clone(), io.clone(), None);
5581
5582 let tx = crate::mvcc::database::LogRecord::for_test(200, &[], None);
5584 let c = log.log_tx(tx).unwrap();
5585 io.wait_for_completion(c).unwrap();
5586
5587 let commit_ts = 201u64;
5590 let db_header = DatabaseHeader::default();
5591 let header_tx = crate::mvcc::database::LogRecord::for_test(commit_ts, &[], Some(db_header));
5592 let c = log.log_tx(header_tx).unwrap();
5593 io.wait_for_completion(c).unwrap();
5594
5595 let mut reader = StreamingLogicalLogReader::new(file.clone(), None);
5596 reader.read_header(&io).unwrap();
5597
5598 let frame = reader
5600 .next_frame_blocking(&io)
5601 .unwrap()
5602 .expect("expected UpdateHeader frame after empty tx");
5603 assert_eq!(frame.len(), 1);
5604 match &frame[0] {
5605 ParsedOp::UpdateHeader {
5606 header: recovered,
5607 commit_ts: recovered_ts,
5608 } => {
5609 assert_eq!(*recovered_ts, commit_ts);
5610 assert_eq!(recovered.magic, db_header.magic);
5611 }
5612 other => panic!("expected UpdateHeader, got {other:?}"),
5613 }
5614
5615 assert!(reader.next_frame_blocking(&io).unwrap().is_none());
5617 }
5618
5619 #[test]
5622 fn test_logical_log_bitflip_integrity_exhaustive_single_frame() {
5623 init_tracing();
5624 let io: Arc<dyn crate::IO> = Arc::new(MemoryIO::new());
5625 let file = io
5626 .open_file("bitflip.db-log", crate::OpenFlags::Create, false)
5627 .unwrap();
5628 let mut log = LogicalLog::new(file.clone(), io.clone(), None);
5629 let mut tx = crate::mvcc::database::LogRecord::new(300);
5630 tx.push_row_version_for_test(&crate::mvcc::database::RowVersion {
5631 id: 1,
5632 begin: crate::mvcc::database::PackedTs::pack(Some(
5633 crate::mvcc::database::TxTimestampOrID::Timestamp(300),
5634 )),
5635 end: crate::mvcc::database::PackedTs::pack(None),
5636 row: generate_simple_string_row((-2).into(), 42, "flip"),
5637 btree_resident: false,
5638 materialized_at: crate::mvcc::database::WalPos::ORIGIN,
5639 });
5640 let c = log.log_tx(tx).unwrap();
5641 io.wait_for_completion(c).unwrap();
5642
5643 let size = file.size().unwrap() as usize;
5644 let mut original = vec![0u8; size];
5645 let read_buf = Arc::new(Buffer::new_temporary(size));
5646 let c = file
5647 .pread(0, Completion::new_read(read_buf.clone(), |_| None))
5648 .unwrap();
5649 io.wait_for_completion(c).unwrap();
5650 original.copy_from_slice(&read_buf.as_slice()[..size]);
5651
5652 for (i, original_byte) in original.iter().enumerate().take(size).skip(LOG_HDR_SIZE) {
5653 for bit in 0..8u8 {
5654 let mutated = original_byte ^ (1 << bit);
5655 let c = file
5656 .pwrite(
5657 i as u64,
5658 Arc::new(Buffer::new(vec![mutated])),
5659 Completion::new_write(|_| {}),
5660 )
5661 .unwrap();
5662 io.wait_for_completion(c).unwrap();
5663
5664 let mut reader = StreamingLogicalLogReader::new(file.clone(), None);
5665 reader.read_header(&io).unwrap();
5666 let res = reader.next_frame_blocking(&io);
5667 match res {
5668 Err(_) | Ok(None) => {}
5669 Ok(Some(frame)) => {
5670 panic!("bit flip at offset={i}, bit={bit} produced valid frame: {frame:?}")
5671 }
5672 }
5673
5674 let c = file
5675 .pwrite(
5676 i as u64,
5677 Arc::new(Buffer::new(vec![*original_byte])),
5678 Completion::new_write(|_| {}),
5679 )
5680 .unwrap();
5681 io.wait_for_completion(c).unwrap();
5682 }
5683 }
5684 }
5685
5686 #[test]
5689 fn test_logical_log_roundtrip_random_table_ops() {
5690 init_tracing();
5691 let seed = 0xA11CE55u64;
5692 let mut rng = ChaCha8Rng::seed_from_u64(seed);
5693 let io: Arc<dyn crate::IO> = Arc::new(MemoryIO::new());
5694 let file = io
5695 .open_file("roundtrip-rand.db-log", crate::OpenFlags::Create, false)
5696 .unwrap();
5697 let mut log = LogicalLog::new(file.clone(), io.clone(), None);
5698
5699 let mut expected = Vec::new();
5700 for tx_i in 0..128u64 {
5701 let mut tx = crate::mvcc::database::LogRecord::new(1_000 + tx_i);
5702 let op_count = (rng.next_u64() % 4) as usize;
5703 for _ in 0..op_count {
5704 let rowid = (rng.next_u64() % 64) as i64 + 1;
5705 let btree_resident = (rng.next_u32() & 1) == 1;
5706 let is_delete = (rng.next_u32() & 1) == 1;
5707 if is_delete {
5708 tx.push_row_version_for_test(&crate::mvcc::database::RowVersion {
5709 id: 0,
5710 begin: crate::mvcc::database::PackedTs::pack(None),
5711 end: crate::mvcc::database::PackedTs::pack(Some(
5712 crate::mvcc::database::TxTimestampOrID::Timestamp(tx.tx_timestamp),
5713 )),
5714 row: Row::new_table_row(
5715 RowID::new((-2).into(), RowKey::Int(rowid)),
5716 &[],
5717 0,
5718 )
5719 .unwrap(),
5720 btree_resident,
5721 materialized_at: crate::mvcc::database::WalPos::ORIGIN,
5722 });
5723 expected.push(ExpectedTableOp::Delete {
5724 rowid,
5725 commit_ts: tx.tx_timestamp,
5726 btree_resident,
5727 });
5728 } else {
5729 let payload = format!("r-{tx_i}-{rowid}");
5730 let row = generate_simple_string_row((-2).into(), rowid, &payload);
5731 tx.push_row_version_for_test(&crate::mvcc::database::RowVersion {
5732 id: 0,
5733 begin: crate::mvcc::database::PackedTs::pack(Some(
5734 crate::mvcc::database::TxTimestampOrID::Timestamp(tx.tx_timestamp),
5735 )),
5736 end: crate::mvcc::database::PackedTs::pack(None),
5737 row: row.clone(),
5738 btree_resident,
5739 materialized_at: crate::mvcc::database::WalPos::ORIGIN,
5740 });
5741 expected.push(ExpectedTableOp::Upsert {
5742 rowid,
5743 payload: row.payload().to_vec(),
5744 commit_ts: tx.tx_timestamp,
5745 btree_resident,
5746 });
5747 }
5748 }
5749 let c = log.log_tx(tx).unwrap();
5750 io.wait_for_completion(c).unwrap();
5751 }
5752
5753 let large_commit_ts = 1_000 + 128u64;
5757 let large_text: String = "x".repeat(200);
5758 let mut large_tx = crate::mvcc::database::LogRecord::new(large_commit_ts);
5759 for rowid in 1..=30i64 {
5760 let row = generate_simple_string_row((-3).into(), rowid, &large_text);
5761 expected.push(ExpectedTableOp::Upsert {
5762 rowid,
5763 payload: row.payload().to_vec(),
5764 commit_ts: large_commit_ts,
5765 btree_resident: false,
5766 });
5767 large_tx.push_row_version_for_test(&crate::mvcc::database::RowVersion {
5768 id: rowid as u64,
5769 begin: crate::mvcc::database::PackedTs::pack(Some(
5770 crate::mvcc::database::TxTimestampOrID::Timestamp(large_commit_ts),
5771 )),
5772 end: crate::mvcc::database::PackedTs::pack(None),
5773 row,
5774 btree_resident: false,
5775 materialized_at: crate::mvcc::database::WalPos::ORIGIN,
5776 });
5777 }
5778 let c = log.log_tx(large_tx).unwrap();
5779 io.wait_for_completion(c).unwrap();
5780
5781 let got = read_table_ops(file.clone(), &io);
5782 assert_eq!(got, expected);
5783 }
5784
5785 #[quickcheck]
5787 fn prop_logical_log_roundtrip_sequence(events: Vec<(bool, i64, bool)>) -> bool {
5788 let io: Arc<dyn crate::IO> = Arc::new(MemoryIO::new());
5789 let file = match io.open_file(
5790 "logical_log_prop_roundtrip_sequence",
5791 OpenFlags::Create,
5792 false,
5793 ) {
5794 Ok(f) => f,
5795 Err(_) => return false,
5796 };
5797 let mut log = LogicalLog::new(file.clone(), io.clone(), None);
5798 let mut expected = Vec::new();
5799
5800 for (idx, (is_delete, rowid, btree_resident)) in events.into_iter().take(64).enumerate() {
5801 let commit_ts = (idx + 1) as u64;
5802 let payload_text = format!("v{idx}");
5803 let row = generate_simple_string_row((-2).into(), rowid, &payload_text);
5804 let row_version = crate::mvcc::database::RowVersion {
5805 id: commit_ts,
5806 begin: crate::mvcc::database::PackedTs::pack(Some(
5807 crate::mvcc::database::TxTimestampOrID::Timestamp(commit_ts),
5808 )),
5809 end: crate::mvcc::database::PackedTs::pack(if is_delete {
5810 Some(crate::mvcc::database::TxTimestampOrID::Timestamp(commit_ts))
5811 } else {
5812 None
5813 }),
5814 row: row.clone(),
5815 btree_resident,
5816 materialized_at: crate::mvcc::database::WalPos::ORIGIN,
5817 };
5818 expected.push(if is_delete {
5819 ExpectedTableOp::Delete {
5820 rowid,
5821 commit_ts,
5822 btree_resident,
5823 }
5824 } else {
5825 ExpectedTableOp::Upsert {
5826 rowid,
5827 payload: row.payload().to_vec(),
5828 commit_ts,
5829 btree_resident,
5830 }
5831 });
5832 let tx = crate::mvcc::database::LogRecord::for_test(commit_ts, &[row_version], None);
5833 let Ok(c) = log.log_tx(tx) else {
5834 return false;
5835 };
5836 if io.wait_for_completion(c).is_err() {
5837 return false;
5838 }
5839 }
5840
5841 if expected.is_empty() {
5842 return file.size().expect("file.size() failed") == 0;
5843 }
5844
5845 read_table_ops(file, &io) == expected
5846 }
5847
5848 #[quickcheck]
5850 fn prop_streaming_varint_roundtrip(value: u64) -> bool {
5851 let mut encoded = [0u8; 9];
5852 let len = write_varint(&mut encoded, value);
5853 if len == 0 || len > 9 {
5854 return false;
5855 }
5856 let encoded = &encoded[..len];
5857
5858 let parsed_streaming = match decode_streaming_varint(encoded) {
5859 Ok(Some(v)) => v,
5860 _ => return false,
5861 };
5862 let parsed_read = match read_varint(encoded) {
5863 Ok(v) => v,
5864 Err(_) => return false,
5865 };
5866
5867 parsed_streaming.0 == value
5868 && parsed_streaming.2 == len
5869 && parsed_streaming.1[..len] == encoded[..]
5870 && parsed_read.0 == value
5871 && parsed_read.1 == len
5872 }
5873
5874 #[quickcheck]
5876 fn prop_streaming_varint_matches_read_varint(bytes: Vec<u8>) -> bool {
5877 let bytes = if bytes.len() > 16 {
5878 &bytes[..16]
5879 } else {
5880 bytes.as_slice()
5881 };
5882 let streaming = decode_streaming_varint(bytes);
5883 let plain = read_varint(bytes);
5884
5885 match (streaming, plain) {
5886 (Ok(Some((v1, b1, l1))), Ok((v2, l2))) => {
5887 v1 == v2 && l1 == l2 && b1[..l1] == bytes[..l1]
5888 }
5889 (Ok(None), Err(_)) => true, (Err(_), Err(_)) => true, _ => false,
5892 }
5893 }
5894
5895 #[test]
5901 fn test_logical_log_btree_resident_roundtrip() {
5902 init_tracing();
5903 let io: Arc<dyn crate::IO> = Arc::new(MemoryIO::new());
5904 let file = io
5905 .open_file("btree.db-log", crate::OpenFlags::Create, false)
5906 .unwrap();
5907 let mut log = LogicalLog::new(file.clone(), io.clone(), None);
5908
5909 let mut tx = crate::mvcc::database::LogRecord::new(55);
5910 let mut row = generate_simple_string_row((-2).into(), 1, "foo");
5911 row.id.table_id = (-2).into();
5912 let version = crate::mvcc::database::RowVersion {
5913 id: 1,
5914 begin: crate::mvcc::database::PackedTs::pack(Some(
5915 crate::mvcc::database::TxTimestampOrID::Timestamp(55),
5916 )),
5917 end: crate::mvcc::database::PackedTs::pack(None),
5918 row,
5919 btree_resident: true,
5920 materialized_at: crate::mvcc::database::WalPos::ORIGIN,
5921 };
5922 tx.push_row_version_for_test(&version);
5923 let c = log.log_tx(tx).unwrap();
5924 io.wait_for_completion(c).unwrap();
5925
5926 let frame_hdr_buf = Arc::new(Buffer::new_temporary(TX_HEADER_SIZE));
5928 let c = file
5929 .pread(
5930 LOG_HDR_SIZE as u64,
5931 Completion::new_read(frame_hdr_buf.clone(), |_| None),
5932 )
5933 .unwrap();
5934 io.wait_for_completion(c).unwrap();
5935 let frame_hdr = frame_hdr_buf.as_slice()[..TX_HEADER_SIZE].to_vec();
5936 assert_eq!(
5937 u32::from_le_bytes(frame_hdr[0..4].try_into().unwrap()),
5938 FRAME_MAGIC,
5939 "FRAME_MAGIC at bytes [0..4]"
5940 );
5941 assert!(
5942 u64::from_le_bytes(frame_hdr[4..12].try_into().unwrap()) > 0,
5943 "payload_size at bytes [4..12] must be non-zero for a non-empty op"
5944 );
5945
5946 let mut reader = StreamingLogicalLogReader::new(file.clone(), None);
5947 reader.read_header(&io).unwrap();
5948 let frame = reader
5949 .next_frame_blocking(&io)
5950 .unwrap()
5951 .expect("expected one frame");
5952 assert_eq!(frame.len(), 1);
5953 match &frame[0] {
5954 ParsedOp::UpsertTable { btree_resident, .. } => {
5955 assert!(*btree_resident);
5956 }
5957 other => panic!("unexpected op: {other:?}"),
5958 }
5959 }
5960
5961 #[test]
5964 fn test_logical_log_header_persistence() {
5965 init_tracing();
5966 let io: Arc<dyn crate::IO> = Arc::new(MemoryIO::new());
5967 let file = io
5968 .open_file("header.db-log", crate::OpenFlags::Create, false)
5969 .unwrap();
5970 let mut log = LogicalLog::new(file.clone(), io.clone(), None);
5971
5972 let mut tx = crate::mvcc::database::LogRecord::new(10);
5973 let row = generate_simple_string_row((-2).into(), 1, "foo");
5974 let version = crate::mvcc::database::RowVersion {
5975 id: 1,
5976 begin: crate::mvcc::database::PackedTs::pack(Some(
5977 crate::mvcc::database::TxTimestampOrID::Timestamp(10),
5978 )),
5979 end: crate::mvcc::database::PackedTs::pack(None),
5980 row,
5981 btree_resident: false,
5982 materialized_at: crate::mvcc::database::WalPos::ORIGIN,
5983 };
5984 tx.push_row_version_for_test(&version);
5985 let c = log.log_tx(tx).unwrap();
5986 io.wait_for_completion(c).unwrap();
5987
5988 let c = file
5989 .truncate(LOG_HDR_SIZE as u64, Completion::new_trunc(|_| {}))
5990 .unwrap();
5991 io.wait_for_completion(c).unwrap();
5992
5993 let mut reader = StreamingLogicalLogReader::new(file.clone(), None);
5994 reader.read_header(&io).unwrap();
5995 let header = reader.header().unwrap();
5996 assert_eq!(header.version, LOG_VERSION_V2);
5997 let encoded = header.encode();
5999 let mut check_buf = [0u8; LOG_HDR_SIZE];
6000 check_buf.copy_from_slice(&encoded);
6001 check_buf[LOG_HDR_CRC_START..LOG_HDR_SIZE].copy_from_slice(&[0; 4]);
6002 let expected_crc = crc32c::crc32c(&check_buf);
6003 assert_eq!(header.hdr_crc32c, expected_crc);
6004 }
6005
6006 #[test]
6009 fn test_logical_log_header_crc_roundtrip() {
6010 init_tracing();
6011 let io: Arc<dyn crate::IO> = Arc::new(MemoryIO::new());
6012 let header = LogHeader::new(&io);
6013 assert_eq!(header.version, LOG_VERSION_V2);
6014 assert_ne!(header.salt, 0, "salt should be non-zero from IO RNG");
6015 let bytes = header.encode();
6016 let mut check_buf = bytes;
6018 check_buf[LOG_HDR_CRC_START..LOG_HDR_SIZE].copy_from_slice(&[0; 4]);
6019 let expected_crc = crc32c::crc32c(&check_buf);
6020 let decoded = LogHeader::decode(&bytes).unwrap();
6021 assert_eq!(decoded.version, header.version);
6022 assert_eq!(decoded.salt, header.salt);
6023 assert_eq!(decoded.hdr_crc32c, expected_crc);
6024 }
6025
6026 #[test]
6029 fn test_try_read_header_reports_invalid_not_corrupt() {
6030 init_tracing();
6031 let io: Arc<dyn crate::IO> = Arc::new(MemoryIO::new());
6032 let file = io
6033 .open_file(
6034 "try-read-header-invalid.db-log",
6035 crate::OpenFlags::Create,
6036 false,
6037 )
6038 .unwrap();
6039 let mut log = LogicalLog::new(file.clone(), io.clone(), None);
6040
6041 append_single_table_op_tx(&mut log, &io, (-2).into(), 1, 11, false, false, "foo");
6042 let c = file
6043 .pwrite(
6044 0,
6045 Arc::new(Buffer::new(vec![0])),
6046 Completion::new_write(|_| {}),
6047 )
6048 .unwrap();
6049 io.wait_for_completion(c).unwrap();
6050
6051 let mut reader = StreamingLogicalLogReader::new(file, None);
6052 let result = reader.try_read_header(&io).unwrap();
6053 assert!(matches!(result, HeaderReadResult::Invalid));
6054 }
6055
6056 #[test]
6060 fn test_truncation_regenerates_salt() {
6061 init_tracing();
6062 let io: Arc<dyn crate::IO> = Arc::new(MemoryIO::new());
6063 let file = io
6064 .open_file("salt-regen.db-log", crate::OpenFlags::Create, false)
6065 .unwrap();
6066 let mut log = LogicalLog::new(file.clone(), io.clone(), None);
6067
6068 append_single_table_op_tx(&mut log, &io, (-2).into(), 1, 10, false, false, "a");
6070 let salt_before = log.header.as_ref().unwrap().salt;
6071
6072 let c = log.truncate(u64::MAX).unwrap();
6076 io.wait_for_completion(c).unwrap();
6077
6078 let salt_after = log.header.as_ref().unwrap().salt;
6079 assert_ne!(salt_before, salt_after, "salt must change on truncation");
6080 assert_eq!(log.offset, 0, "offset must be 0 after truncation");
6081
6082 append_single_table_op_tx(&mut log, &io, (-2).into(), 2, 20, false, false, "b");
6084
6085 let mut reader = StreamingLogicalLogReader::new(file, None);
6087 assert!(matches!(
6088 reader.try_read_header(&io).unwrap(),
6089 HeaderReadResult::Valid(_)
6090 ));
6091 let header = reader.header().unwrap();
6092 assert_eq!(header.salt, salt_after);
6093
6094 match io.block(|| reader.parse_next_transaction()) {
6095 Ok(ParseResult::Frame(frame)) => {
6096 let ops = frame.ops;
6097 assert!(!ops.is_empty(), "expected at least one op");
6098 }
6099 Ok(ParseResult::Eof) => panic!("expected ops, got EOF"),
6100 Ok(ParseResult::InvalidFrame) => panic!("expected ops, got InvalidFrame"),
6101 Err(e) => panic!("expected ops, got error: {e:?}"),
6102 }
6103 assert!(matches!(
6104 io.block(|| reader.parse_next_transaction()),
6105 Ok(ParseResult::Eof)
6106 ));
6107 }
6108
6109 #[test]
6114 fn test_crc_chain_invalidates_suffix_on_corruption() {
6115 init_tracing();
6116 let io: Arc<dyn crate::IO> = Arc::new(MemoryIO::new());
6117 let file = io
6118 .open_file("crc-chain.db-log", crate::OpenFlags::Create, false)
6119 .unwrap();
6120 let mut log = LogicalLog::new(file.clone(), io.clone(), None);
6121
6122 append_single_table_op_tx(&mut log, &io, (-2).into(), 1, 10, false, false, "aaa");
6124 let after_first = log.offset as usize;
6125 append_single_table_op_tx(&mut log, &io, (-2).into(), 2, 20, false, false, "bbb");
6126 append_single_table_op_tx(&mut log, &io, (-2).into(), 3, 30, false, false, "ccc");
6127
6128 let mut reader = StreamingLogicalLogReader::new(file.clone(), None);
6130 assert!(matches!(
6131 reader.try_read_header(&io).unwrap(),
6132 HeaderReadResult::Valid(_)
6133 ));
6134 let mut count = 0;
6135 while let Ok(ParseResult::Frame(_)) = io.block(|| reader.parse_next_transaction()) {
6136 count += 1;
6137 }
6138 assert_eq!(count, 3);
6139
6140 let corrupt_offset = LOG_HDR_SIZE + TX_HEADER_SIZE + 1; let c = file
6143 .pwrite(
6144 corrupt_offset as u64,
6145 Arc::new(Buffer::new(vec![0xFF])),
6146 Completion::new_write(|_| {}),
6147 )
6148 .unwrap();
6149 io.wait_for_completion(c).unwrap();
6150
6151 let mut reader = StreamingLogicalLogReader::new(file, None);
6154 assert!(matches!(
6155 reader.try_read_header(&io).unwrap(),
6156 HeaderReadResult::Valid(_)
6157 ));
6158 match io.block(|| reader.parse_next_transaction()) {
6160 Ok(ParseResult::InvalidFrame) => {}
6161 other => panic!("expected InvalidFrame after corrupted frame 1, got {other:?}"),
6162 }
6163 let valid_offset = reader.last_valid_offset();
6165 assert!(
6166 valid_offset <= after_first,
6167 "valid offset {valid_offset} should be <= first frame end {after_first}",
6168 );
6169 }
6170
6171 #[test]
6177 fn test_splice_frame_from_different_log_rejected() {
6178 init_tracing();
6179 let io: Arc<dyn crate::IO> = Arc::new(MemoryIO::new());
6180
6181 let file_a = io
6183 .open_file("splice-a.db-log", crate::OpenFlags::Create, false)
6184 .unwrap();
6185 let mut log_a = LogicalLog::new(file_a.clone(), io.clone(), None);
6186 append_single_table_op_tx(&mut log_a, &io, (-2).into(), 1, 10, false, false, "aaa");
6187 let log_a_end = log_a.offset as usize;
6188
6189 let file_b = io
6191 .open_file("splice-b.db-log", crate::OpenFlags::Create, false)
6192 .unwrap();
6193 let mut log_b = LogicalLog::new(file_b.clone(), io.clone(), None);
6194 append_single_table_op_tx(&mut log_b, &io, (-2).into(), 2, 20, false, false, "bbb");
6195 let log_b_end = log_b.offset as usize;
6196
6197 let salt_a = log_a.header.as_ref().unwrap().salt;
6199 let salt_b = log_b.header.as_ref().unwrap().salt;
6200 assert_ne!(
6201 salt_a, salt_b,
6202 "two independent logs should have different salts"
6203 );
6204
6205 let frame_b_len = log_b_end - LOG_HDR_SIZE;
6207 let read_buf = Arc::new(Buffer::new_temporary(frame_b_len));
6208 let c = file_b
6209 .pread(
6210 LOG_HDR_SIZE as u64,
6211 Completion::new_read(read_buf.clone(), |_| None),
6212 )
6213 .unwrap();
6214 io.wait_for_completion(c).unwrap();
6215 let frame_b_bytes: Vec<u8> = read_buf.as_slice()[..frame_b_len].to_vec();
6216
6217 let c = file_a
6219 .pwrite(
6220 log_a_end as u64,
6221 Arc::new(Buffer::new(frame_b_bytes)),
6222 Completion::new_write(|_| {}),
6223 )
6224 .unwrap();
6225 io.wait_for_completion(c).unwrap();
6226
6227 let mut reader = StreamingLogicalLogReader::new(file_a, None);
6229 assert!(matches!(
6230 reader.try_read_header(&io).unwrap(),
6231 HeaderReadResult::Valid(_)
6232 ));
6233
6234 match io.block(|| reader.parse_next_transaction()) {
6236 Ok(ParseResult::Frame(frame)) => assert!(!frame.ops.is_empty()),
6237 other => panic!("expected log A's frame to parse, got {other:?}"),
6238 }
6239
6240 match io.block(|| reader.parse_next_transaction()) {
6242 Ok(ParseResult::InvalidFrame) => {}
6243 other => {
6244 panic!("spliced frame from a different log should NOT validate, got {other:?}")
6245 }
6246 }
6247 }
6248
6249 fn test_enc_ctx() -> crate::storage::encryption::EncryptionContext {
6250 use crate::storage::encryption::{CipherMode, EncryptionKey};
6251 let key = EncryptionKey::Key128([0x42u8; 16]);
6252 crate::storage::encryption::EncryptionContext::new(CipherMode::Aes128Gcm, &key, 4096)
6253 .unwrap()
6254 }
6255
6256 fn wrong_key_enc_ctx() -> crate::storage::encryption::EncryptionContext {
6257 use crate::storage::encryption::{CipherMode, EncryptionKey};
6258 let key = EncryptionKey::Key128([0xFFu8; 16]);
6259 crate::storage::encryption::EncryptionContext::new(CipherMode::Aes128Gcm, &key, 4096)
6260 .unwrap()
6261 }
6262
6263 fn make_test_row_version(
6264 table_id: MVTableId,
6265 rowid: i64,
6266 value: &str,
6267 commit_ts: u64,
6268 ) -> crate::mvcc::database::RowVersion {
6269 let row = generate_simple_string_row(table_id, rowid, value);
6270 crate::mvcc::database::RowVersion {
6271 id: rowid as u64,
6272 begin: crate::mvcc::database::PackedTs::pack(Some(
6273 crate::mvcc::database::TxTimestampOrID::Timestamp(commit_ts),
6274 )),
6275 end: crate::mvcc::database::PackedTs::pack(None),
6276 row,
6277 btree_resident: false,
6278 materialized_at: crate::mvcc::database::WalPos::ORIGIN,
6279 }
6280 }
6281
6282 fn make_test_index_row_version(
6283 table_id: MVTableId,
6284 rowid: i64,
6285 value: &str,
6286 commit_ts: u64,
6287 ) -> crate::mvcc::database::RowVersion {
6288 let key_record = ImmutableRecord::from_values(
6289 &[
6290 Value::Text(Text::new(value.to_string())),
6291 Value::from_i64(rowid),
6292 ],
6293 2,
6294 )
6295 .unwrap();
6296 let sortable_key = SortableIndexKey::new_from_record(key_record, test_index_info());
6297 let row_id = RowID::new(table_id, RowKey::Record(Arc::new(sortable_key)));
6298 let row = Row::new_index_row(row_id, 2);
6299 crate::mvcc::database::RowVersion {
6300 id: rowid as u64,
6301 begin: crate::mvcc::database::PackedTs::pack(Some(
6302 crate::mvcc::database::TxTimestampOrID::Timestamp(commit_ts),
6303 )),
6304 end: crate::mvcc::database::PackedTs::pack(None),
6305 row,
6306 btree_resident: false,
6307 materialized_at: crate::mvcc::database::WalPos::ORIGIN,
6308 }
6309 }
6310
6311 fn test_index_info() -> Arc<IndexInfo> {
6312 Arc::new(
6313 IndexInfo::new(
6314 [
6315 crate::types::KeyInfo {
6316 sort_order: turso_parser::ast::SortOrder::Asc,
6317 collation: crate::translate::collate::CollationSeq::Binary,
6318 nulls_order: None,
6319 },
6320 crate::types::KeyInfo {
6321 sort_order: turso_parser::ast::SortOrder::Asc,
6322 collation: crate::translate::collate::CollationSeq::Binary,
6323 nulls_order: None,
6324 },
6325 ],
6326 true,
6327 2,
6328 false,
6329 )
6330 .unwrap(),
6331 )
6332 }
6333
6334 fn make_test_raw_table_row_version(
6335 table_id: MVTableId,
6336 rowid: i64,
6337 record_bytes: Vec<u8>,
6338 commit_ts: u64,
6339 is_delete: bool,
6340 ) -> crate::mvcc::database::RowVersion {
6341 let row =
6342 Row::new_table_row(RowID::new(table_id, RowKey::Int(rowid)), &record_bytes, 1).unwrap();
6343 crate::mvcc::database::RowVersion {
6344 id: rowid as u64,
6345 begin: crate::mvcc::database::PackedTs::pack(if is_delete {
6346 None
6347 } else {
6348 Some(crate::mvcc::database::TxTimestampOrID::Timestamp(commit_ts))
6349 }),
6350 end: crate::mvcc::database::PackedTs::pack(if is_delete {
6351 Some(crate::mvcc::database::TxTimestampOrID::Timestamp(commit_ts))
6352 } else {
6353 None
6354 }),
6355 row,
6356 btree_resident: false,
6357 materialized_at: crate::mvcc::database::WalPos::ORIGIN,
6358 }
6359 }
6360
6361 fn make_test_raw_index_row_version(
6362 table_id: MVTableId,
6363 rowid: i64,
6364 payload_bytes: Vec<u8>,
6365 commit_ts: u64,
6366 is_delete: bool,
6367 ) -> crate::mvcc::database::RowVersion {
6368 let sortable_key = SortableIndexKey::new_from_bytes(payload_bytes, test_index_info());
6369 let row_id = RowID::new(table_id, RowKey::Record(Arc::new(sortable_key)));
6370 let row = Row::new_index_row(row_id, 2);
6371 crate::mvcc::database::RowVersion {
6372 id: rowid as u64,
6373 begin: crate::mvcc::database::PackedTs::pack(if is_delete {
6374 None
6375 } else {
6376 Some(crate::mvcc::database::TxTimestampOrID::Timestamp(commit_ts))
6377 }),
6378 end: crate::mvcc::database::PackedTs::pack(if is_delete {
6379 Some(crate::mvcc::database::TxTimestampOrID::Timestamp(commit_ts))
6380 } else {
6381 None
6382 }),
6383 row,
6384 btree_resident: false,
6385 materialized_at: crate::mvcc::database::WalPos::ORIGIN,
6386 }
6387 }
6388
6389 fn single_upsert_table_op_size_for_text_len(rowid: i64, text_len: usize) -> usize {
6390 let mut encoded = Vec::new();
6391 let value = "x".repeat(text_len);
6392 let row_version = make_test_row_version((-2).into(), rowid, &value, 100);
6393 serialize_op_entry(&mut encoded, &row_version, None).unwrap();
6394 encoded.len()
6395 }
6396
6397 fn try_text_len_for_single_upsert_table_op_size(
6398 rowid: i64,
6399 target_op_size: usize,
6400 ) -> Option<usize> {
6401 (0..=target_op_size).find(|&text_len| {
6402 single_upsert_table_op_size_for_text_len(rowid, text_len) == target_op_size
6403 })
6404 }
6405
6406 fn text_len_for_single_upsert_table_op_size(target_op_size: usize) -> usize {
6407 if let Some(text_len) = try_text_len_for_single_upsert_table_op_size(1, target_op_size) {
6408 return text_len;
6409 }
6410 panic!("could not find text length for op size {target_op_size}");
6411 }
6412
6413 fn try_record_bytes_len_for_upsert_table_op_size(
6414 rowid: i64,
6415 target_op_size: usize,
6416 ) -> Option<usize> {
6417 let rowid_len = varint_len(rowid as u64);
6418 for payload_len_varint_len in 1..=9usize {
6419 let record_bytes_len =
6420 target_op_size.checked_sub(6 + payload_len_varint_len + rowid_len)?;
6421 let payload_len = rowid_len + record_bytes_len;
6422 if varint_len(payload_len as u64) == payload_len_varint_len {
6423 return Some(record_bytes_len);
6424 }
6425 }
6426 None
6427 }
6428
6429 fn read_file_bytes(file: Arc<dyn crate::File>, io: &Arc<dyn crate::IO>) -> Vec<u8> {
6430 let file_size = file.size().unwrap() as usize;
6431 if file_size == 0 {
6432 return Vec::new();
6433 }
6434 let mut reader = StreamingLogicalLogReader::new(file, None);
6435 io.block(|| reader.read_exact_at(0, file_size)).unwrap()
6436 }
6437
6438 fn overwrite_file_bytes(file: Arc<dyn crate::File>, io: &Arc<dyn crate::IO>, bytes: &[u8]) {
6439 let c = file.truncate(0, Completion::new_trunc(|_| {})).unwrap();
6440 io.wait_for_completion(c).unwrap();
6441 if bytes.is_empty() {
6442 return;
6443 }
6444 let c = file
6445 .pwrite(
6446 0,
6447 Arc::new(Buffer::new(bytes.to_vec())),
6448 Completion::new_write(|_| {}),
6449 )
6450 .unwrap();
6451 io.wait_for_completion(c).unwrap();
6452 }
6453
6454 fn open_test_file(io: &Arc<dyn crate::IO>, file_name: &str) -> Arc<dyn crate::File> {
6455 io.open_file(file_name, OpenFlags::Create, false).unwrap()
6456 }
6457
6458 fn append_encrypted_tx(
6459 log: &mut LogicalLog,
6460 io: &Arc<dyn crate::IO>,
6461 tx: crate::mvcc::database::LogRecord,
6462 ) {
6463 let c = log.log_tx(tx).unwrap();
6464 io.wait_for_completion(c).unwrap();
6465 }
6466
6467 fn write_first_encrypted_tx(
6468 file: Arc<dyn crate::File>,
6469 io: &Arc<dyn crate::IO>,
6470 enc_ctx: &crate::storage::encryption::EncryptionContext,
6471 tx: crate::mvcc::database::LogRecord,
6472 ) {
6473 assert_eq!(
6474 file.size().unwrap(),
6475 0,
6476 "write_first_encrypted_tx only supports writing the first frame to a fresh file"
6477 );
6478 let mut log = LogicalLog::new(file, io.clone(), Some(enc_ctx.clone()));
6479 append_encrypted_tx(&mut log, io, tx);
6480 }
6481
6482 fn write_first_encrypted_tx_with_chunk_size_for_test(
6483 file: Arc<dyn crate::File>,
6484 io: &Arc<dyn crate::IO>,
6485 enc_ctx: &crate::storage::encryption::EncryptionContext,
6486 encrypted_payload_chunk_size: usize,
6487 tx: crate::mvcc::database::LogRecord,
6488 ) {
6489 assert_eq!(
6490 file.size().unwrap(),
6491 0,
6492 "write_first_encrypted_tx_with_chunk_size_for_test only supports writing the first frame to a fresh file"
6493 );
6494 let mut log = LogicalLog::new_with_payload_chunk_size(
6495 file,
6496 io.clone(),
6497 Some(enc_ctx.clone()),
6498 encrypted_payload_chunk_size,
6499 );
6500 append_encrypted_tx(&mut log, io, tx);
6501 }
6502
6503 fn write_single_encrypted_tx(
6504 io: &Arc<dyn crate::IO>,
6505 file_name: &str,
6506 enc_ctx: &crate::storage::encryption::EncryptionContext,
6507 tx: crate::mvcc::database::LogRecord,
6508 ) -> Arc<dyn crate::File> {
6509 let file = open_test_file(io, file_name);
6510 write_first_encrypted_tx(file.clone(), io, enc_ctx, tx);
6511 file
6512 }
6513
6514 fn write_single_encrypted_tx_with_chunk_size_for_test(
6515 io: &Arc<dyn crate::IO>,
6516 file_name: &str,
6517 enc_ctx: &crate::storage::encryption::EncryptionContext,
6518 encrypted_payload_chunk_size: usize,
6519 tx: crate::mvcc::database::LogRecord,
6520 ) -> Arc<dyn crate::File> {
6521 let file = open_test_file(io, file_name);
6522 write_first_encrypted_tx_with_chunk_size_for_test(
6523 file.clone(),
6524 io,
6525 enc_ctx,
6526 encrypted_payload_chunk_size,
6527 tx,
6528 );
6529 file
6530 }
6531
6532 fn write_encrypted_txs_with_chunk_size_for_test(
6533 io: &Arc<dyn crate::IO>,
6534 file_name: &str,
6535 enc_ctx: &crate::storage::encryption::EncryptionContext,
6536 encrypted_payload_chunk_size: usize,
6537 txs: Vec<crate::mvcc::database::LogRecord>,
6538 ) -> Arc<dyn crate::File> {
6539 let file = open_test_file(io, file_name);
6540 let mut log = LogicalLog::new_with_payload_chunk_size(
6541 file.clone(),
6542 io.clone(),
6543 Some(enc_ctx.clone()),
6544 encrypted_payload_chunk_size,
6545 );
6546 for tx in txs {
6547 append_encrypted_tx(&mut log, io, tx);
6548 }
6549 file
6550 }
6551
6552 fn parse_only_encrypted_tx_ops(
6553 file: Arc<dyn crate::File>,
6554 io: &Arc<dyn crate::IO>,
6555 enc_ctx: &crate::storage::encryption::EncryptionContext,
6556 ) -> Vec<ParsedOp> {
6557 let mut reader = StreamingLogicalLogReader::new(file, Some(enc_ctx.clone()));
6558 reader.read_header(io).unwrap();
6559 let ops = match io.block(|| reader.parse_next_transaction()).unwrap() {
6560 ParseResult::Frame(frame) => frame.ops,
6561 other => panic!("expected Ops, got {other:?}"),
6562 };
6563 assert!(matches!(
6564 io.block(|| reader.parse_next_transaction()).unwrap(),
6565 ParseResult::Eof
6566 ));
6567 ops
6568 }
6569
6570 fn parse_only_encrypted_tx_ops_with_chunk_size_for_test(
6571 file: Arc<dyn crate::File>,
6572 io: &Arc<dyn crate::IO>,
6573 enc_ctx: &crate::storage::encryption::EncryptionContext,
6574 encrypted_payload_chunk_size: usize,
6575 ) -> Vec<ParsedOp> {
6576 let mut reader = StreamingLogicalLogReader::new_with_payload_chunk_size(
6577 file,
6578 Some(enc_ctx.clone()),
6579 encrypted_payload_chunk_size,
6580 );
6581 reader.read_header(io).unwrap();
6582 let ops = match io.block(|| reader.parse_next_transaction()).unwrap() {
6583 ParseResult::Frame(frame) => frame.ops,
6584 other => panic!("expected Ops, got {other:?}"),
6585 };
6586 assert!(matches!(
6587 io.block(|| reader.parse_next_transaction()).unwrap(),
6588 ParseResult::Eof
6589 ));
6590 ops
6591 }
6592
6593 fn parse_all_encrypted_tx_ops_with_chunk_size_for_test(
6594 file: Arc<dyn crate::File>,
6595 io: &Arc<dyn crate::IO>,
6596 enc_ctx: &crate::storage::encryption::EncryptionContext,
6597 encrypted_payload_chunk_size: usize,
6598 ) -> std::result::Result<Vec<Vec<ParsedOp>>, String> {
6599 let mut reader = StreamingLogicalLogReader::new_with_payload_chunk_size(
6600 file,
6601 Some(enc_ctx.clone()),
6602 encrypted_payload_chunk_size,
6603 );
6604 reader
6605 .read_header(io)
6606 .map_err(|e| format!("failed to read fuzz log header: {e}"))?;
6607 let mut frames = Vec::new();
6608 let mut tx_index = 0usize;
6609 loop {
6610 match io
6611 .block(|| reader.parse_next_transaction())
6612 .map_err(|e| format!("failed to parse fuzz frame {tx_index}: {e}"))?
6613 {
6614 ParseResult::Frame(frame) => frames.push(frame.ops),
6615 ParseResult::Eof => break,
6616 ParseResult::InvalidFrame => {
6617 return Err(format!("invalid fuzz frame at tx_index={tx_index}"));
6618 }
6619 }
6620 tx_index += 1;
6621 }
6622 Ok(frames)
6623 }
6624
6625 fn assert_upsert_table_op(
6626 op: &ParsedOp,
6627 expected_table_id: MVTableId,
6628 expected_rowid: i64,
6629 expected_record_bytes: &[u8],
6630 expected_commit_ts: u64,
6631 ) {
6632 match op {
6633 ParsedOp::UpsertTable {
6634 table_id,
6635 rowid,
6636 record_bytes,
6637 commit_ts,
6638 btree_resident,
6639 } => {
6640 assert_eq!(*table_id, expected_table_id);
6641 assert_eq!(rowid.row_id, RowKey::Int(expected_rowid));
6642 assert_eq!(record_bytes, expected_record_bytes);
6643 assert_eq!(*commit_ts, expected_commit_ts);
6644 assert!(!btree_resident);
6645 }
6646 other => panic!("expected UpsertTable, got {other:?}"),
6647 }
6648 }
6649
6650 fn assert_upsert_index_op(
6651 op: &ParsedOp,
6652 expected_table_id: MVTableId,
6653 expected_payload: &[u8],
6654 expected_commit_ts: u64,
6655 ) {
6656 match op {
6657 ParsedOp::UpsertIndex {
6658 table_id,
6659 payload,
6660 commit_ts,
6661 btree_resident,
6662 } => {
6663 assert_eq!(*table_id, expected_table_id);
6664 assert_eq!(payload, expected_payload);
6665 assert_eq!(*commit_ts, expected_commit_ts);
6666 assert!(!btree_resident);
6667 }
6668 other => panic!("expected UpsertIndex, got {other:?}"),
6669 }
6670 }
6671
6672 fn assert_update_header_op(
6673 op: &ParsedOp,
6674 expected_header: &DatabaseHeader,
6675 expected_commit_ts: u64,
6676 ) {
6677 match op {
6678 ParsedOp::UpdateHeader { header, commit_ts } => {
6679 assert_eq!(*commit_ts, expected_commit_ts);
6680 assert_eq!(
6681 bytemuck::bytes_of(header),
6682 bytemuck::bytes_of(expected_header)
6683 );
6684 }
6685 other => panic!("expected UpdateHeader, got {other:?}"),
6686 }
6687 }
6688
6689 fn encrypted_carry_fuzz_record_bytes_len(
6692 rng: &mut ChaCha8Rng,
6693 rowid: i64,
6694 chunk_size: usize,
6695 ) -> usize {
6696 if rng.random_range(0..4) == 0 {
6698 let exact_op_size = rng.random_range(1..=3) * chunk_size;
6699 if let Some(record_bytes_len) =
6700 try_record_bytes_len_for_upsert_table_op_size(rowid, exact_op_size)
6701 {
6702 return record_bytes_len;
6703 }
6704 }
6705
6706 let jitter = rng.random_range(0..=16) as isize - 8;
6707 let base = match rng.random_range(0..15) {
6708 0 => 1usize,
6709 1 => 16usize,
6710 2 => chunk_size,
6711 3 => chunk_size + 1,
6712 4 => chunk_size - 1,
6713 5 => 2 * chunk_size,
6714 6 => 2 * chunk_size + 1,
6715 7 => 2 * chunk_size - 1,
6716 8 => 3 * chunk_size,
6717 9 => 3 * chunk_size + 1,
6718 10 => chunk_size / 2,
6719 11 => chunk_size + chunk_size / 2,
6720 12 => 2 * chunk_size + chunk_size / 2,
6721 13 => random_range(1..=16usize) + random_range(0..=chunk_size),
6722 14 => random_range(1..=chunk_size),
6723 _ => rng.random_range(1..=3) * chunk_size,
6724 } as isize;
6725 (base + jitter).max(1) as usize
6726 }
6727
6728 fn expected_upsert_table_fuzz_op(
6729 row_version: &crate::mvcc::database::RowVersion,
6730 rowid: i64,
6731 commit_ts: u64,
6732 ) -> ParsedOp {
6733 ParsedOp::UpsertTable {
6734 table_id: (-2).into(),
6735 rowid: RowID::new((-2).into(), RowKey::Int(rowid)),
6736 record_bytes: row_version.row.payload().to_vec(),
6737 commit_ts,
6738 btree_resident: false,
6739 }
6740 }
6741
6742 fn assert_forced_upsert_carry_prefix_layout(
6743 short_filler: &crate::mvcc::database::RowVersion,
6744 short_upsert: &crate::mvcc::database::RowVersion,
6745 long_upsert: &crate::mvcc::database::RowVersion,
6746 chunk_size: usize,
6747 ) {
6748 let mut filler_buf = Vec::new();
6749 serialize_op_entry(&mut filler_buf, short_filler, None).unwrap();
6750 let mut short_upsert_buf = Vec::new();
6751 serialize_op_entry(&mut short_upsert_buf, short_upsert, None).unwrap();
6752 let mut long_upsert_buf = Vec::new();
6753 serialize_op_entry(&mut long_upsert_buf, long_upsert, None).unwrap();
6754
6755 turso_assert_less_than!(
6756 filler_buf.len(),
6757 chunk_size,
6758 "forced short-carry filler upsert must fit before the first chunk boundary"
6759 );
6760 let short_split_offset = chunk_size - filler_buf.len();
6761 turso_assert!(
6762 short_split_offset > 0 && short_split_offset < short_upsert_buf.len(),
6763 "forced short carry must end the first chunk inside the short upsert"
6764 );
6765 turso_assert_less_than!(
6766 short_upsert_buf.len(),
6767 StreamingLogicalLogReader::MAX_SERIALIZED_OP_PREFIX_LEN,
6768 "forced short carry upsert must remain below MAX_SERIALIZED_OP_PREFIX_LEN"
6769 );
6770
6771 let long_start_offset = (filler_buf.len() + short_upsert_buf.len()) % chunk_size;
6772 turso_assert!(
6773 long_start_offset > 0,
6774 "forced long carry upsert must begin inside a chunk, not on a chunk boundary"
6775 );
6776 turso_assert!(
6777 long_upsert_buf.len() > 2 * chunk_size,
6778 "forced long carry upsert must span more than two chunk widths"
6779 );
6780 }
6781
6782 fn append_forced_upsert_carry_prefix(
6783 rng: &mut ChaCha8Rng,
6784 chunk_size: usize,
6785 commit_ts: u64,
6786 row_versions: &mut Vec<crate::mvcc::database::RowVersion>,
6787 expected_ops: &mut Vec<ParsedOp>,
6788 ) {
6789 let short_rowid = 0i64;
6794 let short_record_bytes = vec![0x11];
6795 let short_upsert = make_test_raw_table_row_version(
6796 (-2).into(),
6797 short_rowid,
6798 short_record_bytes,
6799 commit_ts,
6800 false,
6801 );
6802 let mut short_upsert_buf = Vec::new();
6803 serialize_op_entry(&mut short_upsert_buf, &short_upsert, None).unwrap();
6804 turso_assert_less_than!(
6805 short_upsert_buf.len(),
6806 StreamingLogicalLogReader::MAX_SERIALIZED_OP_PREFIX_LEN,
6807 "forced short carry upsert must remain below MAX_SERIALIZED_OP_PREFIX_LEN"
6808 );
6809
6810 let split_offset = rng.random_range(1..short_upsert_buf.len());
6811 let filler_op_size = chunk_size - split_offset;
6812 let filler_record_bytes_len =
6813 try_record_bytes_len_for_upsert_table_op_size(1, filler_op_size)
6814 .expect("forced filler upsert size must map to a valid record_bytes length");
6815 let short_filler = make_test_raw_table_row_version(
6816 (-2).into(),
6817 1,
6818 vec![0x22; filler_record_bytes_len],
6819 commit_ts,
6820 false,
6821 );
6822 let long_upsert = make_test_raw_table_row_version(
6823 (-2).into(),
6824 2,
6825 vec![0x5A; 2 * chunk_size + rng.random_range(64..=256)],
6826 commit_ts,
6827 false,
6828 );
6829 assert_forced_upsert_carry_prefix_layout(
6830 &short_filler,
6831 &short_upsert,
6832 &long_upsert,
6833 chunk_size,
6834 );
6835
6836 expected_ops.push(expected_upsert_table_fuzz_op(&short_filler, 1, commit_ts));
6837 row_versions.push(short_filler);
6838
6839 expected_ops.push(expected_upsert_table_fuzz_op(
6840 &short_upsert,
6841 short_rowid,
6842 commit_ts,
6843 ));
6844 row_versions.push(short_upsert);
6845
6846 expected_ops.push(expected_upsert_table_fuzz_op(&long_upsert, 2, commit_ts));
6847 row_versions.push(long_upsert);
6848 }
6849
6850 fn generate_random_encrypted_carry_fuzz_upsert(
6851 rng: &mut ChaCha8Rng,
6852 rowid: i64,
6853 chunk_size: usize,
6854 commit_ts: u64,
6855 ) -> (crate::mvcc::database::RowVersion, ParsedOp) {
6856 let record_bytes_len = encrypted_carry_fuzz_record_bytes_len(rng, rowid, chunk_size);
6858 let row_version = make_test_raw_table_row_version(
6859 (-2).into(),
6860 rowid,
6861 vec![(rowid as u8).wrapping_add(1); record_bytes_len],
6862 commit_ts,
6863 false,
6864 );
6865 let expected = expected_upsert_table_fuzz_op(&row_version, rowid, commit_ts);
6866 (row_version, expected)
6867 }
6868
6869 fn generate_encrypted_carry_fuzz_case(
6871 case_seed: u64,
6872 chunk_size: usize,
6873 include_forced_prefix: bool,
6874 ) -> (Vec<crate::mvcc::database::LogRecord>, Vec<Vec<ParsedOp>>) {
6875 let mut rng = ChaCha8Rng::seed_from_u64(case_seed);
6876 let tx_count = rng.random_range(1..=3);
6877 let mut txs = Vec::with_capacity(tx_count);
6878 let mut expected_frames = Vec::with_capacity(tx_count);
6879
6880 for tx_index in 0..tx_count {
6881 let commit_ts = 1_000 + (rng.next_u64() % 1_000_000) + tx_index as u64;
6882 let op_count = rng.random_range(1..=20);
6883
6884 let mut row_versions = Vec::with_capacity(op_count);
6885 let mut expected_ops = Vec::with_capacity(op_count);
6886 if tx_index == 0 && include_forced_prefix {
6890 append_forced_upsert_carry_prefix(
6891 &mut rng,
6892 chunk_size,
6893 commit_ts,
6894 &mut row_versions,
6895 &mut expected_ops,
6896 );
6897 }
6898
6899 while row_versions.len() < op_count {
6900 let rowid = (row_versions.len() + 1) as i64;
6901 let (row_version, expected_op) = generate_random_encrypted_carry_fuzz_upsert(
6902 &mut rng, rowid, chunk_size, commit_ts,
6903 );
6904 row_versions.push(row_version);
6905 expected_ops.push(expected_op);
6906 }
6907
6908 txs.push(crate::mvcc::database::LogRecord::for_test(
6909 commit_ts,
6910 &row_versions,
6911 None,
6912 ));
6913 expected_frames.push(expected_ops);
6914 }
6915
6916 (txs, expected_frames)
6917 }
6918
6919 fn encrypted_chunk_ranges(
6922 payload_size: usize,
6923 tag_size: usize,
6924 nonce_size: usize,
6925 ) -> Vec<std::ops::Range<usize>> {
6926 let mut ranges = Vec::new();
6927 let mut offset = 0usize;
6928 for chunk_index in
6929 0..encrypted_payload_chunk_count(payload_size, ENCRYPTED_PAYLOAD_CHUNK_SIZE)
6930 {
6931 let plaintext_len = encrypted_chunk_plaintext_len(
6932 payload_size,
6933 chunk_index,
6934 ENCRYPTED_PAYLOAD_CHUNK_SIZE,
6935 )
6936 .unwrap();
6937 let chunk_len = encrypted_chunk_blob_size(plaintext_len, tag_size, nonce_size).unwrap();
6938 ranges.push(offset..offset + chunk_len);
6939 offset += chunk_len;
6940 }
6941 ranges
6942 }
6943
6944 fn assert_single_frame_invalid(
6945 file: Arc<dyn crate::File>,
6946 io: &Arc<dyn crate::IO>,
6947 enc_ctx: crate::storage::encryption::EncryptionContext,
6948 ) {
6949 let mut reader = StreamingLogicalLogReader::new(file, Some(enc_ctx));
6950 reader.read_header(io).unwrap();
6951 match io.block(|| reader.parse_next_transaction()).unwrap() {
6952 ParseResult::InvalidFrame => {}
6953 other => panic!("expected InvalidFrame, got {other:?}"),
6954 }
6955 }
6956
6957 #[test]
6961 fn test_encrypted_log_roundtrip_and_layout() {
6962 init_tracing();
6963 let io: Arc<dyn crate::IO> = Arc::new(MemoryIO::new());
6964 let file = open_test_file(&io, "enc-roundtrip.db-log");
6965 let table_id: MVTableId = (-2).into();
6966 let enc_ctx = test_enc_ctx();
6967 let tag_size = enc_ctx.tag_size();
6968 let nonce_size = enc_ctx.nonce_size();
6969 let expected_hello_record_bytes = generate_simple_string_row(table_id, 1, "hello")
6970 .payload()
6971 .to_vec();
6972 let expected_world_record_bytes = generate_simple_string_row(table_id, 2, "world")
6973 .payload()
6974 .to_vec();
6975
6976 let tx = crate::mvcc::database::LogRecord::for_test(
6978 100,
6979 &[
6980 make_test_row_version(table_id, 1, "hello", 100),
6981 make_test_row_version(table_id, 2, "world", 100),
6982 ],
6983 None,
6984 );
6985 write_first_encrypted_tx(file.clone(), &io, &enc_ctx, tx);
6986
6987 let frame_hdr_buf = Arc::new(Buffer::new_temporary(TX_HEADER_SIZE));
6990 let frame_hdr_out = Arc::new(crate::sync::RwLock::new(Vec::new()));
6991 let out = frame_hdr_out.clone();
6992 let c = Completion::new_read(
6993 frame_hdr_buf,
6994 Box::new(
6995 move |res: std::result::Result<(Arc<Buffer>, i32), crate::CompletionError>| {
6996 let Ok((buf, n)) = res else { return None };
6997 out.write().extend_from_slice(&buf.as_slice()[..n as usize]);
6998 None
6999 },
7000 ),
7001 );
7002 let c = file.pread(LOG_HDR_SIZE as u64, c).unwrap();
7003 io.wait_for_completion(c).unwrap();
7004
7005 let frame_hdr = frame_hdr_out.read();
7006 assert_eq!(frame_hdr.len(), TX_HEADER_SIZE);
7007 let payload_size = u64::from_le_bytes(frame_hdr[4..12].try_into().unwrap()) as usize;
7008
7009 let file_size = file.size().unwrap() as usize;
7010 let encrypted_blob_size = file_size - LOG_HDR_SIZE - TX_HEADER_SIZE - TX_TRAILER_SIZE;
7011 let expected_blob_size = encrypted_payload_blob_size(
7012 payload_size,
7013 ENCRYPTED_PAYLOAD_CHUNK_SIZE,
7014 tag_size,
7015 nonce_size,
7016 )
7017 .unwrap();
7018 assert_eq!(
7019 encrypted_blob_size, expected_blob_size,
7020 "on-disk blob size ({encrypted_blob_size}) != expected chunked encrypted size({expected_blob_size})"
7021 );
7022
7023 let mut reader = StreamingLogicalLogReader::new(file, Some(enc_ctx));
7025 reader.read_header(&io).unwrap();
7026
7027 let ops = match io.block(|| reader.parse_next_transaction()).unwrap() {
7028 ParseResult::Frame(frame) => frame.ops,
7029 other => panic!("expected Ops, got {other:?}"),
7030 };
7031 assert_eq!(ops.len(), 2);
7032 assert_upsert_table_op(&ops[0], table_id, 1, &expected_hello_record_bytes, 100);
7033 assert_upsert_table_op(&ops[1], table_id, 2, &expected_world_record_bytes, 100);
7034
7035 assert!(matches!(
7036 io.block(|| reader.parse_next_transaction()).unwrap(),
7037 ParseResult::Eof
7038 ));
7039 }
7040
7041 #[test]
7045 fn test_encrypted_log_roundtrip_with_test_chunk_size_override() {
7046 init_tracing();
7047 let io: Arc<dyn crate::IO> = Arc::new(MemoryIO::new());
7048 let enc_ctx = test_enc_ctx();
7049 const TEST_CHUNK_SIZE: usize = 2 * 1024;
7050
7051 let target_op_size = TEST_CHUNK_SIZE + 257;
7052 let text_len = text_len_for_single_upsert_table_op_size(target_op_size);
7053 let value = "t".repeat(text_len);
7054 let row_version = make_test_row_version((-2).into(), 1, &value, 100);
7055 let expected_record_bytes = row_version.row.payload().to_vec();
7056 let tx = crate::mvcc::database::LogRecord::for_test(100, &[row_version], None);
7057
7058 let file = write_single_encrypted_tx_with_chunk_size_for_test(
7059 &io,
7060 "enc-roundtrip-test-chunk-size.db-log",
7061 &enc_ctx,
7062 TEST_CHUNK_SIZE,
7063 tx,
7064 );
7065
7066 assert_eq!(
7067 encrypted_payload_chunk_count(target_op_size, TEST_CHUNK_SIZE),
7068 2,
7069 "test payload should span exactly two test-sized chunks"
7070 );
7071 let expected_blob_size = encrypted_payload_blob_size(
7072 target_op_size,
7073 TEST_CHUNK_SIZE,
7074 enc_ctx.tag_size(),
7075 enc_ctx.nonce_size(),
7076 )
7077 .unwrap();
7078 assert_eq!(
7079 file.size().unwrap() as usize,
7080 LOG_HDR_SIZE + TX_HEADER_SIZE + expected_blob_size + TX_TRAILER_SIZE
7081 );
7082
7083 let ops = parse_only_encrypted_tx_ops_with_chunk_size_for_test(
7084 file,
7085 &io,
7086 &enc_ctx,
7087 TEST_CHUNK_SIZE,
7088 );
7089 assert_eq!(ops.len(), 1);
7090 assert_upsert_table_op(&ops[0], (-2).into(), 1, &expected_record_bytes, 100);
7091 }
7092
7093 #[test]
7096 fn test_encrypted_log_carry_fuzz() {
7097 init_tracing();
7098 let io: Arc<dyn crate::IO> = Arc::new(MemoryIO::new());
7099 let enc_ctx = test_enc_ctx();
7100 const TEST_CHUNK_SIZE: usize = 2 * 1024;
7101
7102 let seed = std::env::var("TURSO_ENCRYPTED_CARRY_FUZZ_SEED")
7103 .ok()
7104 .and_then(|value| value.parse::<u64>().ok())
7105 .unwrap_or_else(|| rng().random::<u64>());
7106 let mut rng = ChaCha8Rng::seed_from_u64(seed);
7107 let case_count = rng.random_range(1..=8);
7108 let forced_case_index = rng.random_range(0..case_count);
7109 eprintln!(
7110 "encrypted carry fuzz root_seed={seed} case_count={case_count} forced_case_index={forced_case_index} test_chunk_size={TEST_CHUNK_SIZE}"
7111 );
7112
7113 for case_index in 0..case_count {
7114 let case_seed = rng.next_u64();
7115 let include_forced_prefix = case_index == forced_case_index;
7116 let (txs, expected_frames) = generate_encrypted_carry_fuzz_case(
7117 case_seed,
7118 TEST_CHUNK_SIZE,
7119 include_forced_prefix,
7120 );
7121
7122 let file = write_encrypted_txs_with_chunk_size_for_test(
7123 &io,
7124 &format!("enc-carry-fuzz-{seed}-{case_index}.db-log"),
7125 &enc_ctx,
7126 TEST_CHUNK_SIZE,
7127 txs,
7128 );
7129 let actual_frames = parse_all_encrypted_tx_ops_with_chunk_size_for_test(
7130 file,
7131 &io,
7132 &enc_ctx,
7133 TEST_CHUNK_SIZE,
7134 )
7135 .unwrap_or_else(|err| {
7136 panic!(
7137 "encrypted carry fuzz failed while parsing frames: root_seed={seed} case_index={case_index} forced_case_index={forced_case_index} include_forced_prefix={include_forced_prefix} case_seed={case_seed} err={err}"
7138 )
7139 });
7140
7141 assert_eq!(
7142 actual_frames, expected_frames,
7143 "encrypted carry fuzz failed: root_seed={seed} case_index={case_index} forced_case_index={forced_case_index} include_forced_prefix={include_forced_prefix} case_seed={case_seed}"
7144 );
7145 }
7146 }
7147
7148 #[test]
7149 fn test_encrypted_log_format_assumptions_are_pinned() {
7150 assert_eq!(LOG_VERSION_V2, 2);
7151 assert_eq!(LOG_VERSION, 3);
7152 assert_eq!(LOG_HDR_SIZE, 56);
7153 assert_eq!(ENCRYPTED_PAYLOAD_CHUNK_SIZE, 32 * 1024);
7154 assert_eq!(ENCRYPTED_CHUNK_AAD_SIZE, 32);
7155 assert_eq!(FRAME_MAGIC, 0x5854_564D);
7156 assert_eq!(EXT_FRAME_MAGIC, 0x5845_564D);
7157 assert_eq!(END_MAGIC, 0x4554_564D);
7158 assert_eq!(TX_HEADER_SIZE_V2, 24);
7159 assert_eq!(TX_HEADER_SIZE, 24);
7160 assert_eq!(TX_EXT_HEADER_SIZE, 40);
7161 assert_eq!(TX_TRAILER_SIZE, 8);
7162 }
7163
7164 #[test]
7165 fn test_non_portable_first_write_uses_lml2_header_and_v2_frame() {
7166 init_tracing();
7167 let io: Arc<dyn crate::IO> = Arc::new(MemoryIO::new());
7168 let file = io
7169 .open_file(
7170 "non-portable-first-write-lml2.db-log",
7171 OpenFlags::Create,
7172 false,
7173 )
7174 .unwrap();
7175 let mut log = LogicalLog::new(file.clone(), io.clone(), None);
7176
7177 let tx = crate::mvcc::database::LogRecord::for_test(
7178 10,
7179 &[make_test_row_version((-2).into(), 1, "visible", 10)],
7180 None,
7181 );
7182 let c = log.log_tx(tx).unwrap();
7183 io.wait_for_completion(c).unwrap();
7184
7185 let frame = read_file_bytes(file, &io);
7186 let header = LogHeader::decode(&frame[..LOG_HDR_SIZE]).unwrap();
7187 assert_eq!(header.version, LOG_VERSION_V2);
7188 assert_eq!(
7189 u32::from_le_bytes(frame[LOG_HDR_SIZE..LOG_HDR_SIZE + 4].try_into().unwrap()),
7190 FRAME_MAGIC
7191 );
7192 }
7193
7194 #[test]
7195 fn test_non_portable_appends_keep_lml2_header_and_v2_frames() {
7196 init_tracing();
7197 let io: Arc<dyn crate::IO> = Arc::new(MemoryIO::new());
7198 let file = io
7199 .open_file("non-portable-appends-lml2.db-log", OpenFlags::Create, false)
7200 .unwrap();
7201 let mut log = LogicalLog::new(file.clone(), io.clone(), None);
7202
7203 for (commit_ts, rowid) in [(10, 1), (20, 2)] {
7204 let tx = crate::mvcc::database::LogRecord::for_test(
7205 commit_ts,
7206 &[make_test_row_version(
7207 (-2).into(),
7208 rowid,
7209 "visible",
7210 commit_ts,
7211 )],
7212 None,
7213 );
7214 let c = log.log_tx(tx).unwrap();
7215 io.wait_for_completion(c).unwrap();
7216 }
7217
7218 let frame = read_file_bytes(file, &io);
7219 let header = LogHeader::decode(&frame[..LOG_HDR_SIZE]).unwrap();
7220 assert_eq!(header.version, LOG_VERSION_V2);
7221 assert_eq!(
7222 u32::from_le_bytes(frame[LOG_HDR_SIZE..LOG_HDR_SIZE + 4].try_into().unwrap()),
7223 FRAME_MAGIC
7224 );
7225
7226 let first_payload_size = u64::from_le_bytes(
7227 frame[LOG_HDR_SIZE + 4..LOG_HDR_SIZE + 12]
7228 .try_into()
7229 .unwrap(),
7230 ) as usize;
7231 let second_frame_start =
7232 LOG_HDR_SIZE + TX_HEADER_SIZE + first_payload_size + TX_TRAILER_SIZE;
7233 assert_eq!(
7234 u32::from_le_bytes(
7235 frame[second_frame_start..second_frame_start + 4]
7236 .try_into()
7237 .unwrap()
7238 ),
7239 FRAME_MAGIC
7240 );
7241 }
7242
7243 #[cfg(clt_turso_feature = "conn_raw_api")]
7244 #[test]
7245 fn test_portable_changes_upgrade_non_empty_lml2_log_to_lml3() {
7246 init_tracing();
7247 let io: Arc<dyn crate::IO> = Arc::new(MemoryIO::new());
7248 let file = io
7249 .open_file(
7250 "portable-after-lml2-upgrade.db-log",
7251 OpenFlags::Create,
7252 false,
7253 )
7254 .unwrap();
7255 let mut log = LogicalLog::new(file.clone(), io.clone(), None);
7256
7257 let tx = crate::mvcc::database::LogRecord::for_test(
7258 10,
7259 &[make_test_row_version((-2).into(), 1, "visible", 10)],
7260 None,
7261 );
7262 let c = log.log_tx(tx).unwrap();
7263 io.wait_for_completion(c).unwrap();
7264
7265 let mut portable_tx = crate::mvcc::database::LogRecord::for_test(
7266 20,
7267 &[make_test_row_version((-2).into(), 2, "visible", 20)],
7268 None,
7269 );
7270 portable_tx.portable_changes_enabled = true;
7271 portable_tx.portable_changes = vec![0x1a, 0x00];
7272
7273 let c = log
7274 .upgrade_header_for_log_tx(&portable_tx)
7275 .unwrap()
7276 .unwrap();
7277 io.wait_for_completion(c).unwrap();
7278 let c = log.log_tx(portable_tx).unwrap();
7279 io.wait_for_completion(c).unwrap();
7280
7281 let frame = read_file_bytes(file, &io);
7282 let header = LogHeader::decode(&frame[..LOG_HDR_SIZE]).unwrap();
7283 assert_eq!(header.version, LOG_VERSION);
7284
7285 let first_payload_size = u64::from_le_bytes(
7286 frame[LOG_HDR_SIZE + 4..LOG_HDR_SIZE + 12]
7287 .try_into()
7288 .unwrap(),
7289 ) as usize;
7290 let second_frame_start =
7291 LOG_HDR_SIZE + TX_HEADER_SIZE + first_payload_size + TX_TRAILER_SIZE;
7292 assert_eq!(
7293 u32::from_le_bytes(
7294 frame[second_frame_start..second_frame_start + 4]
7295 .try_into()
7296 .unwrap()
7297 ),
7298 EXT_FRAME_MAGIC
7299 );
7300 }
7301
7302 #[cfg(clt_turso_feature = "conn_raw_api")]
7303 #[test]
7304 fn test_next_portable_change_frame_returns_empty_and_nonempty_lml3_frames() {
7305 init_tracing();
7306 let io: Arc<dyn crate::IO> = Arc::new(MemoryIO::new());
7307 let file = io
7308 .open_file(
7309 "sync-frame-empty-and-nonempty.db-log",
7310 OpenFlags::Create,
7311 false,
7312 )
7313 .unwrap();
7314 let mut log = LogicalLog::new(file.clone(), io.clone(), None);
7315
7316 let mut empty_sync_tx = crate::mvcc::database::LogRecord::for_test(
7317 10,
7318 &[make_test_row_version((-2).into(), 1, "internal", 10)],
7319 None,
7320 );
7321 empty_sync_tx.portable_changes_enabled = true;
7322 let c = log.log_tx(empty_sync_tx).unwrap();
7323 io.wait_for_completion(c).unwrap();
7324
7325 let encoded_empty_logical_op = vec![0x1a, 0x00];
7326 let mut sync_tx = crate::mvcc::database::LogRecord::for_test(
7327 20,
7328 &[make_test_row_version((-2).into(), 2, "visible", 20)],
7329 None,
7330 );
7331 sync_tx.portable_changes = encoded_empty_logical_op;
7332 let c = log.log_tx(sync_tx).unwrap();
7333 io.wait_for_completion(c).unwrap();
7334
7335 let mut reader = StreamingLogicalLogReader::new(file, None);
7336 reader.read_header(&io).unwrap();
7337 assert_eq!(reader.header().unwrap().version, LOG_VERSION);
7338 let first = io
7339 .block(|| reader.next_portable_change_frame())
7340 .unwrap()
7341 .unwrap();
7342 assert_eq!(first.commit_ts, 10);
7343 assert_eq!(first.extension_record_count, 0);
7344 assert!(first.payload.is_empty());
7345
7346 let second = io
7347 .block(|| reader.next_portable_change_frame())
7348 .unwrap()
7349 .unwrap();
7350 assert_eq!(second.commit_ts, 20);
7351 assert_eq!(second.extension_record_count, 1);
7352 assert!(!second.payload.is_empty());
7353 assert_eq!(second.end_offset, reader.last_valid_offset() as u64);
7354
7355 assert!(io
7356 .block(|| reader.next_portable_change_frame())
7357 .unwrap()
7358 .is_none());
7359 }
7360
7361 #[cfg(clt_turso_feature = "conn_raw_api")]
7362 #[test]
7363 fn test_portable_extension_block_precedes_recovery_payload() {
7364 init_tracing();
7365 let io: Arc<dyn crate::IO> = Arc::new(MemoryIO::new());
7366 let file = io
7367 .open_file(
7368 "portable-extension-before-payload.db-log",
7369 OpenFlags::Create,
7370 false,
7371 )
7372 .unwrap();
7373 let mut log = LogicalLog::new(file.clone(), io.clone(), None);
7374
7375 let portable_metadata = vec![0x1a, 0x00];
7376 let mut tx = crate::mvcc::database::LogRecord::for_test(
7377 20,
7378 &[make_test_row_version((-2).into(), 2, "visible", 20)],
7379 None,
7380 );
7381 tx.portable_changes = portable_metadata.clone();
7382 let c = log.log_tx(tx).unwrap();
7383 io.wait_for_completion(c).unwrap();
7384
7385 let frame = read_file_bytes(file, &io);
7386 let tx_header_start = LOG_HDR_SIZE;
7387 let body_start = LOG_HDR_SIZE + TX_EXT_HEADER_SIZE;
7388 assert_eq!(
7389 u32::from_le_bytes(
7390 frame[tx_header_start..tx_header_start + 4]
7391 .try_into()
7392 .unwrap()
7393 ),
7394 EXT_FRAME_MAGIC
7395 );
7396 let extension_size = u64::from_le_bytes(
7397 frame[tx_header_start + 24..tx_header_start + 32]
7398 .try_into()
7399 .unwrap(),
7400 ) as usize;
7401 assert!(extension_size >= EXTENSION_RECORD_HEADER_SIZE);
7402
7403 let extension_type =
7404 u16::from_le_bytes(frame[body_start..body_start + 2].try_into().unwrap());
7405 assert_eq!(extension_type, EXTENSION_TYPE_PORTABLE_CHANGES);
7406 let extension_payload_len = u32::from_le_bytes(
7407 frame[body_start + 4..body_start + EXTENSION_RECORD_HEADER_SIZE]
7408 .try_into()
7409 .unwrap(),
7410 ) as usize;
7411 let extension_payload = &frame[body_start + EXTENSION_RECORD_HEADER_SIZE
7412 ..body_start + EXTENSION_RECORD_HEADER_SIZE + extension_payload_len];
7413 assert!(extension_payload.ends_with(&portable_metadata));
7414
7415 let recovery_start = body_start + extension_size;
7416 assert_eq!(frame[recovery_start], OP_UPSERT_TABLE);
7417 }
7418
7419 #[test]
7420 fn test_next_portable_change_frame_does_not_advance_lml2_logs() {
7421 init_tracing();
7422 let io: Arc<dyn crate::IO> = Arc::new(MemoryIO::new());
7423 let file = io
7424 .open_file("sync-frame-lml2.db-log", OpenFlags::Create, false)
7425 .unwrap();
7426
7427 let mut header = LogHeader::new(&io);
7428 header.version = LOG_VERSION_V2;
7429 let buffer = Arc::new(Buffer::new(header.encode().to_vec()));
7430 let c = Completion::new_write(|_| {});
7431 io.wait_for_completion(file.pwrite(0, buffer, c).unwrap())
7432 .unwrap();
7433
7434 let mut reader = StreamingLogicalLogReader::new(file, None);
7435 reader.read_header(&io).unwrap();
7436 assert_eq!(reader.last_valid_offset(), LOG_HDR_SIZE);
7437 assert!(io
7438 .block(|| reader.next_portable_change_frame())
7439 .unwrap()
7440 .is_none());
7441 assert_eq!(reader.last_valid_offset(), LOG_HDR_SIZE);
7442 }
7443
7444 #[test]
7445 fn test_encrypted_chunk_aad_layout_is_pinned() {
7446 let non_last_aad = build_encrypted_chunk_aad(
7447 0x0102_0304_0506_0708,
7448 None,
7449 0x2122_2324,
7450 0x3132_3334_3536_3738,
7451 0x4142_4344,
7452 );
7453 assert_eq!(
7454 non_last_aad,
7455 [
7456 0x08, 0x07, 0x06, 0x05, 0x04, 0x03, 0x02, 0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
7458 0x00, 0x24, 0x23, 0x22, 0x21, 0x38, 0x37, 0x36, 0x35, 0x34, 0x33, 0x32, 0x31, 0x44, 0x43, 0x42, 0x41, ]
7463 );
7464
7465 let last_aad = build_encrypted_chunk_aad(
7466 0x0102_0304_0506_0708,
7467 Some(0x1112_1314_1516_1718),
7468 0x2122_2324,
7469 0x3132_3334_3536_3738,
7470 0x4142_4344,
7471 );
7472
7473 assert_eq!(
7474 last_aad,
7475 [
7476 0x08, 0x07, 0x06, 0x05, 0x04, 0x03, 0x02, 0x01, 0x18, 0x17, 0x16, 0x15, 0x14, 0x13, 0x12,
7478 0x11, 0x24, 0x23, 0x22, 0x21, 0x38, 0x37, 0x36, 0x35, 0x34, 0x33, 0x32, 0x31, 0x44, 0x43, 0x42, 0x41, ]
7483 );
7484 }
7485
7486 #[test]
7487 fn test_encrypted_log_aes128_chunk_layout_assumptions_are_pinned() {
7488 init_tracing();
7489 let io: Arc<dyn crate::IO> = Arc::new(MemoryIO::new());
7490 let enc_ctx = test_enc_ctx();
7491
7492 assert_eq!(
7493 enc_ctx.cipher_mode(),
7494 crate::storage::encryption::CipherMode::Aes128Gcm
7495 );
7496 assert_eq!(enc_ctx.tag_size(), 16);
7497 assert_eq!(enc_ctx.nonce_size(), 12);
7498
7499 for (payload_size, expected_chunk_ranges, expected_file_size) in [
7500 (
7501 32_767usize,
7502 std::iter::once(0..32_795).collect::<Vec<_>>(),
7503 32_883usize,
7504 ),
7505 (
7506 32_768usize,
7507 std::iter::once(0..32_796).collect::<Vec<_>>(),
7508 32_884usize,
7509 ),
7510 (32_769usize, vec![0..32_796, 32_796..32_825], 32_913usize),
7511 (65_536usize, vec![0..32_796, 32_796..65_592], 65_680usize),
7512 (
7513 65_537usize,
7514 vec![0..32_796, 32_796..65_592, 65_592..65_621],
7515 65_709usize,
7516 ),
7517 ] {
7518 let text_len = text_len_for_single_upsert_table_op_size(payload_size);
7519 let value = "p".repeat(text_len);
7520 let tx = crate::mvcc::database::LogRecord::for_test(
7521 100,
7522 &[make_test_row_version((-2).into(), 1, &value, 100)],
7523 None,
7524 );
7525 let file = write_single_encrypted_tx(
7526 &io,
7527 &format!("enc-layout-pinned-{payload_size}.db-log"),
7528 &enc_ctx,
7529 tx,
7530 );
7531
7532 let frame_bytes = read_file_bytes(file.clone(), &io);
7533 let actual_payload_size = u64::from_le_bytes(
7534 frame_bytes[LOG_HDR_SIZE + 4..LOG_HDR_SIZE + 12]
7535 .try_into()
7536 .unwrap(),
7537 ) as usize;
7538 assert_eq!(actual_payload_size, payload_size);
7539 assert_eq!(file.size().unwrap() as usize, expected_file_size);
7540 assert_eq!(
7541 encrypted_chunk_ranges(payload_size, 16, 12),
7542 expected_chunk_ranges
7543 );
7544 }
7545 }
7546
7547 #[test]
7550 fn test_encrypted_log_payload_size_tamper_rejected() {
7551 init_tracing();
7552 let io: Arc<dyn crate::IO> = Arc::new(MemoryIO::new());
7553 let file = io
7554 .open_file("enc-payload-size-tamper.db-log", OpenFlags::Create, false)
7555 .unwrap();
7556 let enc_ctx = test_enc_ctx();
7557 let table_id: MVTableId = (-2).into();
7558 let text_len =
7559 text_len_for_single_upsert_table_op_size(2 * ENCRYPTED_PAYLOAD_CHUNK_SIZE + 257);
7560 let value = "s".repeat(text_len);
7561
7562 let mut log = LogicalLog::new(file.clone(), io.clone(), Some(enc_ctx.clone()));
7563 let tx = crate::mvcc::database::LogRecord::for_test(
7564 444,
7565 &[make_test_row_version(table_id, 1, &value, 444)],
7566 None,
7567 );
7568 append_encrypted_tx(&mut log, &io, tx);
7569
7570 let frame_bytes = read_file_bytes(file.clone(), &io);
7571 let payload_size = u64::from_le_bytes(
7572 frame_bytes[LOG_HDR_SIZE + 4..LOG_HDR_SIZE + 12]
7573 .try_into()
7574 .unwrap(),
7575 );
7576 let bad_payload_size = Arc::new(Buffer::new((payload_size + 1).to_le_bytes().to_vec()));
7577 let c = Completion::new_write(|_| {});
7578 io.wait_for_completion(
7579 file.pwrite((LOG_HDR_SIZE + 4) as u64, bad_payload_size, c)
7580 .unwrap(),
7581 )
7582 .unwrap();
7583
7584 let mut reader = StreamingLogicalLogReader::new(file, Some(enc_ctx));
7585 reader.read_header(&io).unwrap();
7586 match io.block(|| reader.parse_next_transaction()).unwrap() {
7587 ParseResult::InvalidFrame => {}
7588 other => panic!("expected InvalidFrame after payload_size tamper, got {other:?}"),
7589 }
7590 }
7591
7592 #[test]
7593 fn test_encrypted_log_chunk_layout_boundaries() {
7594 init_tracing();
7595 let io: Arc<dyn crate::IO> = Arc::new(MemoryIO::new());
7596 let enc_ctx = test_enc_ctx();
7597 let tag_size = enc_ctx.tag_size();
7598 let nonce_size = enc_ctx.nonce_size();
7599
7600 for target_op_size in [
7601 ENCRYPTED_PAYLOAD_CHUNK_SIZE - 1,
7602 ENCRYPTED_PAYLOAD_CHUNK_SIZE,
7603 ENCRYPTED_PAYLOAD_CHUNK_SIZE + 1,
7604 2 * ENCRYPTED_PAYLOAD_CHUNK_SIZE,
7605 2 * ENCRYPTED_PAYLOAD_CHUNK_SIZE + 1,
7606 ] {
7607 let text_len = text_len_for_single_upsert_table_op_size(target_op_size);
7608 let value = "x".repeat(text_len);
7609 let row_version = make_test_row_version((-2).into(), 1, &value, 100);
7610 let expected_record_bytes = row_version.row.payload().to_vec();
7611 let tx = crate::mvcc::database::LogRecord::for_test(100, &[row_version], None);
7612 let file = write_single_encrypted_tx(
7613 &io,
7614 &format!("enc-layout-{target_op_size}.db-log"),
7615 &enc_ctx,
7616 tx,
7617 );
7618
7619 let frame_hdr = read_file_bytes(file.clone(), &io);
7620 let payload_size = u64::from_le_bytes(
7621 frame_hdr[LOG_HDR_SIZE + 4..LOG_HDR_SIZE + 12]
7622 .try_into()
7623 .unwrap(),
7624 ) as usize;
7625 assert_eq!(payload_size, target_op_size);
7626
7627 let file_size = file.size().unwrap() as usize;
7628 let encrypted_blob_size = file_size - LOG_HDR_SIZE - TX_HEADER_SIZE - TX_TRAILER_SIZE;
7629 let expected_blob_size = encrypted_payload_blob_size(
7630 payload_size,
7631 ENCRYPTED_PAYLOAD_CHUNK_SIZE,
7632 tag_size,
7633 nonce_size,
7634 )
7635 .unwrap();
7636 assert_eq!(encrypted_blob_size, expected_blob_size);
7637
7638 let ops = parse_only_encrypted_tx_ops(file, &io, &enc_ctx);
7639 assert_eq!(ops.len(), 1);
7640 assert_upsert_table_op(&ops[0], (-2).into(), 1, &expected_record_bytes, 100);
7641 }
7642 }
7643
7644 #[test]
7645 fn test_encrypted_log_single_op_crosses_chunk_boundary() {
7646 init_tracing();
7647 let io: Arc<dyn crate::IO> = Arc::new(MemoryIO::new());
7648 let enc_ctx = test_enc_ctx();
7649 let target_op_size = ENCRYPTED_PAYLOAD_CHUNK_SIZE + 257;
7650 let text_len = text_len_for_single_upsert_table_op_size(target_op_size);
7651 let value = "x".repeat(text_len);
7652 let row_version = make_test_row_version((-2).into(), 1, &value, 100);
7653 let expected_record_bytes = row_version.row.payload().to_vec();
7654
7655 let tx = crate::mvcc::database::LogRecord::for_test(100, &[row_version], None);
7656 let file = write_single_encrypted_tx(&io, "enc-cross-boundary.db-log", &enc_ctx, tx);
7657 let ops = parse_only_encrypted_tx_ops(file, &io, &enc_ctx);
7658 assert_eq!(ops.len(), 1);
7659 assert_upsert_table_op(&ops[0], (-2).into(), 1, &expected_record_bytes, 100);
7660 }
7661
7662 #[test]
7665 fn test_encrypted_log_varint_crosses_chunk_boundary() {
7666 init_tracing();
7667 let io: Arc<dyn crate::IO> = Arc::new(MemoryIO::new());
7668 let file = io
7669 .open_file("enc-varint-boundary.db-log", OpenFlags::Create, false)
7670 .unwrap();
7671 let enc_ctx = test_enc_ctx();
7672 let filler_len = text_len_for_single_upsert_table_op_size(ENCRYPTED_PAYLOAD_CHUNK_SIZE - 7);
7676 let filler_value = "a".repeat(filler_len);
7677 let second_value = "b".repeat(200);
7678 let filler = make_test_row_version((-2).into(), 1, &filler_value, 100);
7679 let second = make_test_row_version((-2).into(), 2, &second_value, 100);
7680 let expected_filler_record_bytes = filler.row.payload().to_vec();
7681 let expected_second_record_bytes = second.row.payload().to_vec();
7682
7683 let mut filler_buf = Vec::new();
7684 serialize_op_entry(&mut filler_buf, &filler, None).unwrap();
7685 assert_eq!(filler_buf.len(), ENCRYPTED_PAYLOAD_CHUNK_SIZE - 7);
7686
7687 let mut second_buf = Vec::new();
7688 serialize_op_entry(&mut second_buf, &second, None).unwrap();
7689 let (_, varint_bytes) = read_varint_partial(&second_buf[6..]).unwrap().unwrap();
7693 assert!(
7694 varint_bytes >= 2,
7695 "second op payload_len must use a multi-byte varint so the chunk boundary can split it"
7696 );
7697 assert_eq!(
7701 filler_buf.len() + 6 + 1,
7702 ENCRYPTED_PAYLOAD_CHUNK_SIZE,
7703 "chunk boundary should fall after the first payload_len varint byte"
7704 );
7705
7706 let tx = crate::mvcc::database::LogRecord::for_test(100, &[filler, second], None);
7707 write_first_encrypted_tx(file.clone(), &io, &enc_ctx, tx);
7708 let ops = parse_only_encrypted_tx_ops(file, &io, &enc_ctx);
7709 assert_eq!(ops.len(), 2);
7710 assert_upsert_table_op(&ops[0], (-2).into(), 1, &expected_filler_record_bytes, 100);
7711 assert_upsert_table_op(&ops[1], (-2).into(), 2, &expected_second_record_bytes, 100);
7712 }
7713
7714 #[test]
7717 fn test_encrypted_log_header_op_crosses_chunk_boundary() {
7718 init_tracing();
7719 let io: Arc<dyn crate::IO> = Arc::new(MemoryIO::new());
7720 let file = io
7721 .open_file("enc-header-boundary.db-log", OpenFlags::Create, false)
7722 .unwrap();
7723 let enc_ctx = test_enc_ctx();
7724 let mut header_buf = Vec::new();
7725 let mut header = DatabaseHeader::default();
7726 header.database_size = 123.into();
7727 header.schema_cookie = 456.into();
7728 serialize_header_entry(&mut header_buf, &header);
7729
7730 let filler_payload_size = ENCRYPTED_PAYLOAD_CHUNK_SIZE - (header_buf.len() - 1);
7731 let filler_len = text_len_for_single_upsert_table_op_size(filler_payload_size);
7732 let filler_value = "h".repeat(filler_len);
7733 let filler = make_test_row_version((-2).into(), 1, &filler_value, 100);
7734 let expected_filler_record_bytes = filler.row.payload().to_vec();
7735
7736 let mut filler_buf = Vec::new();
7737 serialize_op_entry(&mut filler_buf, &filler, None).unwrap();
7738 assert_eq!(filler_buf.len(), filler_payload_size);
7739 assert_eq!(
7740 filler_buf.len() + header_buf.len() - 1,
7741 ENCRYPTED_PAYLOAD_CHUNK_SIZE,
7742 "chunk boundary should split the header op after its first byte"
7743 );
7744
7745 let tx = crate::mvcc::database::LogRecord::for_test(100, &[filler], Some(header));
7746 write_first_encrypted_tx(file.clone(), &io, &enc_ctx, tx);
7747 let ops = parse_only_encrypted_tx_ops(file, &io, &enc_ctx);
7748 assert_eq!(ops.len(), 2);
7749 assert_upsert_table_op(&ops[0], (-2).into(), 1, &expected_filler_record_bytes, 100);
7750 assert_update_header_op(&ops[1], &header, 100);
7751 }
7752
7753 #[test]
7756 fn test_encrypted_log_many_ops_cross_chunk_boundaries() {
7757 init_tracing();
7758 let io: Arc<dyn crate::IO> = Arc::new(MemoryIO::new());
7759 let file = io
7760 .open_file("enc-many-ops.db-log", OpenFlags::Create, false)
7761 .unwrap();
7762 let enc_ctx = test_enc_ctx();
7763 let table_id: MVTableId = (-2).into();
7764
7765 let row_versions = (0..96)
7766 .map(|rowid| {
7767 let value = format!("row-{rowid}-{}", "x".repeat(900));
7768 make_test_row_version(table_id, rowid + 1, &value, 200)
7769 })
7770 .collect::<Vec<_>>();
7771 let expected_record_bytes = row_versions
7772 .iter()
7773 .map(|row_version| row_version.row.payload().to_vec())
7774 .collect::<Vec<_>>();
7775 let tx = crate::mvcc::database::LogRecord::for_test(200, &row_versions, None);
7776 write_first_encrypted_tx(file.clone(), &io, &enc_ctx, tx);
7777 let ops = parse_only_encrypted_tx_ops(file, &io, &enc_ctx);
7778 assert_eq!(ops.len(), 96);
7779 for (idx, op) in ops.iter().enumerate() {
7780 assert_upsert_table_op(
7781 op,
7782 table_id,
7783 (idx + 1) as i64,
7784 &expected_record_bytes[idx],
7785 200,
7786 );
7787 }
7788 }
7789
7790 #[test]
7793 fn test_encrypted_log_upsert_index_crosses_chunk_boundary() {
7794 init_tracing();
7795 let io: Arc<dyn crate::IO> = Arc::new(MemoryIO::new());
7796 let enc_ctx = test_enc_ctx();
7797 let index_id: MVTableId = (-3).into();
7798 let value = "i".repeat(ENCRYPTED_PAYLOAD_CHUNK_SIZE * 2);
7799 let row_version = make_test_index_row_version(index_id, 42, &value, 250);
7800 let expected_payload = row_version.row.payload().to_vec();
7801
7802 let tx = crate::mvcc::database::LogRecord::for_test(250, &[row_version], None);
7803 let file = write_single_encrypted_tx(&io, "enc-index-boundary.db-log", &enc_ctx, tx);
7804
7805 let frame_bytes = read_file_bytes(file.clone(), &io);
7806 let payload_size = u64::from_le_bytes(
7807 frame_bytes[LOG_HDR_SIZE + 4..LOG_HDR_SIZE + 12]
7808 .try_into()
7809 .unwrap(),
7810 ) as usize;
7811 assert!(
7812 payload_size > ENCRYPTED_PAYLOAD_CHUNK_SIZE,
7813 "index payload should span multiple encrypted chunks"
7814 );
7815
7816 let ops = parse_only_encrypted_tx_ops(file, &io, &enc_ctx);
7817 assert_eq!(ops.len(), 1);
7818 assert_upsert_index_op(&ops[0], index_id, &expected_payload, 250);
7819 }
7820
7821 #[test]
7824 fn test_encrypted_log_multiple_frames_crc_chain() {
7825 init_tracing();
7826 let io: Arc<dyn crate::IO> = Arc::new(MemoryIO::new());
7827 let file = io
7828 .open_file("enc-multi.db-log", OpenFlags::Create, false)
7829 .unwrap();
7830 let table_id: MVTableId = (-2).into();
7831 let enc_ctx = test_enc_ctx();
7832 let expected_record_bytes = (0..5u64)
7833 .map(|i| generate_simple_string_row(table_id, i as i64, &format!("val_{i}")))
7834 .map(|row| row.payload().to_vec())
7835 .collect::<Vec<_>>();
7836
7837 let mut log = LogicalLog::new(file.clone(), io.clone(), Some(enc_ctx.clone()));
7838 for i in 0..5u64 {
7839 let tx = crate::mvcc::database::LogRecord::for_test(
7840 100 + i,
7841 &[make_test_row_version(
7842 table_id,
7843 i as i64,
7844 &format!("val_{i}"),
7845 100 + i,
7846 )],
7847 None,
7848 );
7849 append_encrypted_tx(&mut log, &io, tx);
7850 }
7851
7852 let mut reader = StreamingLogicalLogReader::new(file, Some(enc_ctx));
7853 reader.read_header(&io).unwrap();
7854
7855 for i in 0..5u64 {
7856 let ops = match io.block(|| reader.parse_next_transaction()).unwrap() {
7857 ParseResult::Frame(frame) => frame.ops,
7858 other => panic!("frame {i}: expected Ops, got {other:?}"),
7859 };
7860 assert_eq!(ops.len(), 1, "frame {i}");
7861 assert_upsert_table_op(
7862 &ops[0],
7863 table_id,
7864 i as i64,
7865 &expected_record_bytes[i as usize],
7866 100 + i,
7867 );
7868 }
7869
7870 assert!(matches!(
7871 io.block(|| reader.parse_next_transaction()).unwrap(),
7872 ParseResult::Eof
7873 ));
7874 }
7875
7876 #[test]
7878 fn test_encrypted_log_integrity_rejection() {
7879 init_tracing();
7880 let table_id: MVTableId = (-2).into();
7881 let enc_ctx = test_enc_ctx();
7882
7883 {
7885 let io: Arc<dyn crate::IO> = Arc::new(MemoryIO::new());
7886 let file = io
7887 .open_file("enc-wrongkey.db-log", OpenFlags::Create, false)
7888 .unwrap();
7889
7890 let mut log = LogicalLog::new(file.clone(), io.clone(), Some(enc_ctx.clone()));
7891 let tx = crate::mvcc::database::LogRecord::for_test(
7892 100,
7893 &[make_test_row_version(table_id, 1, "secret", 100)],
7894 None,
7895 );
7896 append_encrypted_tx(&mut log, &io, tx);
7897
7898 let mut reader = StreamingLogicalLogReader::new(file, Some(wrong_key_enc_ctx()));
7899 reader.read_header(&io).unwrap();
7900
7901 match io.block(|| reader.parse_next_transaction()).unwrap() {
7902 ParseResult::InvalidFrame => {}
7903 other => panic!("expected InvalidFrame with wrong key, got {other:?}"),
7904 }
7905 }
7906
7907 {
7911 let io: Arc<dyn crate::IO> = Arc::new(MemoryIO::new());
7912 let file = io
7913 .open_file("enc-hdr-tamper.db-log", OpenFlags::Create, false)
7914 .unwrap();
7915
7916 let mut log = LogicalLog::new(file.clone(), io.clone(), Some(enc_ctx.clone()));
7917 let tx = crate::mvcc::database::LogRecord::for_test(
7918 100,
7919 &[make_test_row_version(table_id, 1, "hdr_tamper", 100)],
7920 None,
7921 );
7922 append_encrypted_tx(&mut log, &io, tx);
7923
7924 let corrupt_offset = (LOG_HDR_SIZE + 16) as u64;
7926 let byte_buf = Arc::new(Buffer::new(vec![0xFF]));
7927 let c = Completion::new_write(move |_| {});
7928 io.wait_for_completion(file.pwrite(corrupt_offset, byte_buf, c).unwrap())
7929 .unwrap();
7930
7931 let mut reader = StreamingLogicalLogReader::new(file, Some(enc_ctx.clone()));
7932 reader.read_header(&io).unwrap();
7933
7934 match io.block(|| reader.parse_next_transaction()).unwrap() {
7935 ParseResult::InvalidFrame => {}
7936 other => panic!("expected InvalidFrame after TX header tamper, got {other:?}"),
7937 }
7938 }
7939
7940 {
7942 let io: Arc<dyn crate::IO> = Arc::new(MemoryIO::new());
7943 let file = io
7944 .open_file("enc-tamper.db-log", OpenFlags::Create, false)
7945 .unwrap();
7946
7947 let mut log = LogicalLog::new(file.clone(), io.clone(), Some(enc_ctx.clone()));
7948 let tx = crate::mvcc::database::LogRecord::for_test(
7949 100,
7950 &[make_test_row_version(table_id, 1, "tamper_me", 100)],
7951 None,
7952 );
7953 append_encrypted_tx(&mut log, &io, tx);
7954
7955 let corrupt_offset = (LOG_HDR_SIZE + TX_HEADER_SIZE + 1) as u64;
7962 let pattern: Vec<u8> = (0..16)
7963 .map(|i| if i & 1 == 0 { 0x00 } else { 0xFF })
7964 .collect();
7965 let byte_buf = Arc::new(Buffer::new(pattern));
7966 let c = Completion::new_write(move |_| {});
7967 io.wait_for_completion(file.pwrite(corrupt_offset, byte_buf, c).unwrap())
7968 .unwrap();
7969
7970 let mut reader = StreamingLogicalLogReader::new(file, Some(enc_ctx));
7971 reader.read_header(&io).unwrap();
7972
7973 match io.block(|| reader.parse_next_transaction()).unwrap() {
7974 ParseResult::InvalidFrame => {}
7975 other => panic!("expected InvalidFrame after ciphertext tamper, got {other:?}"),
7976 }
7977 }
7978 }
7979
7980 #[test]
7983 fn test_encrypted_log_torn_tail_rejected() {
7984 init_tracing();
7985 let io: Arc<dyn crate::IO> = Arc::new(MemoryIO::new());
7986 let file = open_test_file(&io, "enc-torn.db-log");
7987 let table_id: MVTableId = (-2).into();
7988 let enc_ctx = test_enc_ctx();
7989 let first_row_version = make_test_row_version(table_id, 0, "data", 100);
7990 let expected_first_record_bytes = first_row_version.row.payload().to_vec();
7991
7992 let mut log = LogicalLog::new(file.clone(), io.clone(), Some(enc_ctx.clone()));
7994 let first_tx = crate::mvcc::database::LogRecord::for_test(100, &[first_row_version], None);
7995 append_encrypted_tx(&mut log, &io, first_tx);
7996 let second_tx = crate::mvcc::database::LogRecord::for_test(
7997 101,
7998 &[make_test_row_version(table_id, 1, "data", 101)],
7999 None,
8000 );
8001 append_encrypted_tx(&mut log, &io, second_tx);
8002
8003 let file_size = file.size().unwrap();
8005 let truncate_at = file_size - 5; let c = Completion::new_trunc(|_| {});
8007 io.wait_for_completion(file.truncate(truncate_at, c).unwrap())
8008 .unwrap();
8009
8010 let mut reader = StreamingLogicalLogReader::new(file, Some(enc_ctx));
8011 reader.read_header(&io).unwrap();
8012
8013 match io.block(|| reader.parse_next_transaction()).unwrap() {
8015 ParseResult::Frame(frame) => {
8016 let ops = frame.ops;
8017 assert_eq!(ops.len(), 1);
8018 assert_upsert_table_op(&ops[0], (-2).into(), 0, &expected_first_record_bytes, 100);
8019 }
8020 other => panic!("expected Ops for frame 1, got {other:?}"),
8021 }
8022
8023 match io.block(|| reader.parse_next_transaction()).unwrap() {
8025 ParseResult::Eof => {}
8026 other => panic!("expected Eof for torn frame 2, got {other:?}"),
8027 }
8028 }
8029
8030 #[test]
8033 fn test_encrypted_log_chunk_integrity_rejection() {
8034 init_tracing();
8035 let io: Arc<dyn crate::IO> = Arc::new(MemoryIO::new());
8036 let enc_ctx = test_enc_ctx();
8037 let text_len =
8038 text_len_for_single_upsert_table_op_size(2 * ENCRYPTED_PAYLOAD_CHUNK_SIZE + 257);
8039 let value = "z".repeat(text_len);
8040
8041 let base_file = open_test_file(&io, "enc-chunk-integrity-base.db-log");
8042 let row_version = make_test_row_version((-2).into(), 1, &value, 333);
8043 let expected_record_bytes = row_version.row.payload().to_vec();
8044 let tx = crate::mvcc::database::LogRecord::for_test(333, &[row_version], None);
8045 write_first_encrypted_tx(base_file.clone(), &io, &enc_ctx, tx);
8046 let base_ops = parse_only_encrypted_tx_ops(base_file.clone(), &io, &enc_ctx);
8047 assert_eq!(base_ops.len(), 1);
8048 assert_upsert_table_op(&base_ops[0], (-2).into(), 1, &expected_record_bytes, 333);
8049
8050 let base_bytes = read_file_bytes(base_file, &io);
8051 let payload_size = u64::from_le_bytes(
8052 base_bytes[LOG_HDR_SIZE + 4..LOG_HDR_SIZE + 12]
8053 .try_into()
8054 .unwrap(),
8055 ) as usize;
8056 let chunk_ranges =
8057 encrypted_chunk_ranges(payload_size, enc_ctx.tag_size(), enc_ctx.nonce_size());
8058 assert!(
8059 chunk_ranges.len() >= 3,
8060 "expected at least 3 encrypted chunks for corruption coverage"
8061 );
8062 let frame_payload_start = LOG_HDR_SIZE + TX_HEADER_SIZE;
8063 let full_chunk_plaintext_len =
8064 encrypted_chunk_plaintext_len(payload_size, 1, ENCRYPTED_PAYLOAD_CHUNK_SIZE).unwrap();
8065
8066 let mut cases: Vec<(&str, Vec<u8>, bool)> = Vec::new();
8067
8068 {
8070 let mut bytes = base_bytes.clone();
8071 let offset = frame_payload_start + chunk_ranges[1].start + 1;
8072 bytes[offset] ^= 0xFF;
8073 cases.push(("ciphertext", bytes, false));
8074 }
8075
8076 {
8078 let mut bytes = base_bytes.clone();
8079 let offset = frame_payload_start + chunk_ranges[1].start + full_chunk_plaintext_len + 1;
8080 bytes[offset] ^= 0xFF;
8081 cases.push(("tag", bytes, false));
8082 }
8083
8084 {
8086 let mut bytes = base_bytes.clone();
8087 let offset = frame_payload_start
8088 + chunk_ranges[1].start
8089 + full_chunk_plaintext_len
8090 + enc_ctx.tag_size();
8091 bytes[offset] ^= 0xFF;
8092 cases.push(("nonce", bytes, false));
8093 }
8094
8095 {
8097 let mut bytes = base_bytes.clone();
8098 let first = chunk_ranges[0].clone();
8099 let second = chunk_ranges[1].clone();
8100 let first_bytes =
8101 bytes[frame_payload_start + first.start..frame_payload_start + first.end].to_vec();
8102 let second_bytes = bytes
8103 [frame_payload_start + second.start..frame_payload_start + second.end]
8104 .to_vec();
8105 bytes[frame_payload_start + first.start..frame_payload_start + first.end]
8106 .copy_from_slice(&second_bytes);
8107 bytes[frame_payload_start + second.start..frame_payload_start + second.end]
8108 .copy_from_slice(&first_bytes);
8109 cases.push(("reorder", bytes, false));
8110 }
8111
8112 {
8114 let mut bytes = base_bytes.clone();
8115 let second = chunk_ranges[1].clone();
8116 bytes.drain(frame_payload_start + second.start..frame_payload_start + second.end);
8117 cases.push(("drop", bytes, true));
8118 }
8119
8120 {
8122 let mut bytes = base_bytes;
8123 let first = chunk_ranges[0].clone();
8124 let second = chunk_ranges[1].clone();
8125 let first_bytes =
8126 bytes[frame_payload_start + first.start..frame_payload_start + first.end].to_vec();
8127 bytes[frame_payload_start + second.start..frame_payload_start + second.end]
8128 .copy_from_slice(&first_bytes);
8129 cases.push(("duplicate", bytes, false));
8130 }
8131
8132 for (label, bytes, allow_eof) in cases {
8133 let file = io
8134 .open_file(
8135 &format!("enc-chunk-integrity-{label}.db-log"),
8136 OpenFlags::Create,
8137 false,
8138 )
8139 .unwrap();
8140 overwrite_file_bytes(file.clone(), &io, &bytes);
8141 if allow_eof {
8142 let mut reader = StreamingLogicalLogReader::new(file, Some(enc_ctx.clone()));
8143 reader.read_header(&io).unwrap();
8144 match io.block(|| reader.parse_next_transaction()).unwrap() {
8145 ParseResult::InvalidFrame | ParseResult::Eof => {}
8146 other => panic!("expected rejection for {label}, got {other:?}"),
8147 }
8148 } else {
8149 assert_single_frame_invalid(file, &io, enc_ctx.clone());
8150 }
8151 }
8152 }
8153
8154 #[test]
8157 fn test_encrypted_log_chunk_torn_tail_rejected() {
8158 init_tracing();
8159 let io: Arc<dyn crate::IO> = Arc::new(MemoryIO::new());
8160 let file = io
8161 .open_file("enc-chunk-torn-tail.db-log", OpenFlags::Create, false)
8162 .unwrap();
8163 let enc_ctx = test_enc_ctx();
8164 let table_id: MVTableId = (-2).into();
8165 let text_len =
8166 text_len_for_single_upsert_table_op_size(2 * ENCRYPTED_PAYLOAD_CHUNK_SIZE + 257);
8167 let value = "q".repeat(text_len);
8168
8169 let mut log = LogicalLog::new(file.clone(), io.clone(), Some(enc_ctx.clone()));
8170 let first_row_version = make_test_row_version(table_id, 1, "prefix", 500);
8171 let expected_prefix_record_bytes = first_row_version.row.payload().to_vec();
8172 let first_tx = crate::mvcc::database::LogRecord::for_test(500, &[first_row_version], None);
8173 let c = log.log_tx(first_tx).unwrap();
8174 io.wait_for_completion(c).unwrap();
8175 let second_frame_start = log.offset as usize;
8176
8177 let second_tx = crate::mvcc::database::LogRecord::for_test(
8178 600,
8179 &[make_test_row_version(table_id, 2, &value, 600)],
8180 None,
8181 );
8182 let c = log.log_tx(second_tx).unwrap();
8183 io.wait_for_completion(c).unwrap();
8184
8185 let base_bytes = read_file_bytes(file.clone(), &io);
8186 let second_payload_size = u64::from_le_bytes(
8187 base_bytes[second_frame_start + 4..second_frame_start + 12]
8188 .try_into()
8189 .unwrap(),
8190 ) as usize;
8191 let chunk_ranges = encrypted_chunk_ranges(
8192 second_payload_size,
8193 enc_ctx.tag_size(),
8194 enc_ctx.nonce_size(),
8195 );
8196 assert!(chunk_ranges.len() >= 3);
8197 let second_payload_start = second_frame_start + TX_HEADER_SIZE;
8198 let second_chunk_plaintext_len =
8199 encrypted_chunk_plaintext_len(second_payload_size, 1, ENCRYPTED_PAYLOAD_CHUNK_SIZE)
8200 .unwrap();
8201 let second_chunk = chunk_ranges[1].clone();
8202 let last_chunk = chunk_ranges.last().unwrap().clone();
8203 let second_frame_end = base_bytes.len();
8204
8205 let cuts = [
8206 second_payload_start + second_chunk.start + 17,
8207 second_payload_start
8208 + second_chunk.start
8209 + second_chunk_plaintext_len
8210 + enc_ctx.tag_size(),
8211 second_payload_start + second_chunk.end,
8212 second_frame_end - TX_TRAILER_SIZE + 3,
8213 ];
8214
8215 for (idx, cut) in cuts.into_iter().enumerate() {
8216 let file = io
8217 .open_file(
8218 &format!("enc-chunk-torn-tail-{idx}.db-log"),
8219 OpenFlags::Create,
8220 false,
8221 )
8222 .unwrap();
8223 overwrite_file_bytes(file.clone(), &io, &base_bytes[..cut]);
8224
8225 let mut reader = StreamingLogicalLogReader::new(file, Some(enc_ctx.clone()));
8226 reader.read_header(&io).unwrap();
8227 match io.block(|| reader.parse_next_transaction()).unwrap() {
8228 ParseResult::Frame(frame) => {
8229 let ops = frame.ops;
8230 assert_eq!(ops.len(), 1);
8231 assert_upsert_table_op(
8232 &ops[0],
8233 (-2).into(),
8234 1,
8235 &expected_prefix_record_bytes,
8236 500,
8237 );
8238 }
8239 other => panic!("expected prefix frame to survive, got {other:?}"),
8240 }
8241 match io.block(|| reader.parse_next_transaction()).unwrap() {
8242 ParseResult::Eof => {}
8243 other => panic!("expected Eof for torn multi-chunk frame, got {other:?}"),
8244 }
8245 }
8246
8247 assert!(last_chunk.end > last_chunk.start);
8249 }
8250}