[{"_1":2,"_3":-5,"_4":-5},"loaderData",{"_5":6,"_7":8,"_9":10,"_11":12},"actionData","errors","root",{"_179":-5,"_182":183,"_184":185},"routes/shell",{"_41":42,"_178":-5,"_179":-5,"_180":181},"routes/repo",{"_34":-7,"_35":36,"_37":38,"_39":40,"_41":42,"_43":32,"_44":32,"_45":32,"_46":32,"_47":32,"_48":32,"_49":50,"_51":50,"_52":32,"_53":-7,"_54":-7,"_55":56,"_57":58,"_59":60,"_61":-5,"_62":63,"_64":65,"_66":67,"_68":69},"routes/repo.blob",{"_13":14,"_15":16,"_17":18,"_19":20,"_21":22,"_23":24,"_25":-5,"_26":27,"_28":29,"_29":30,"_31":32,"_33":14},"ref","main","path","crates/ltx/src/codec.rs","oid","d69775c711703fff874bc369d31f389af5ba5a4e","size",15001,"commit","90b43017241f81189453d326d05948f388b34652","mounts",[],"source","plain",{"_13":14,"_15":16},"kind","text","//! Streaming LTX v0.5.2 encoder and backward-compatible decoder.\n\nuse crate::error::{Error, Result};\nuse crate::ltx::{\n checksum_page, lock_pgno, Crc64, Header, PageHeader, Trailer, CHECKSUM_SIZE, HEADER_SIZE,\n PAGE_HEADER_FLAG_SIZE, PAGE_HEADER_SIZE, TRAILER_SIZE,\n};\nuse crate::CHECKSUM_FLAG;\nuse std::collections::BTreeMap;\nuse std::io::{Read, Write};\n\n#[derive(Debug, Clone, Copy, PartialEq, Eq)]\nenum DecoderState {\n Header,\n Pages,\n Close,\n Closed,\n}\n\n#[derive(Debug, Clone, Copy, PartialEq, Eq)]\nenum PageEncoding {\n LegacyFrame,\n SizedBlock,\n}\n\npub(crate) struct Decoder {\n reader: R,\n state: DecoderState,\n pub(crate) header: Header,\n pub(crate) trailer: Trailer,\n hash: Crc64,\n rolling_checksum: u64,\n}\n\nimpl Decoder {\n pub(crate) fn new(reader: R) -> Self {\n Self {\n reader,\n state: DecoderState::Header,\n header: Header::default(),\n trailer: Trailer::default(),\n hash: Crc64::new(),\n rolling_checksum: 0,\n }\n }\n\n pub(crate) fn decode_header(&mut self) -> Result<()> {\n if self.state != DecoderState::Header {\n return Err(Error::LTXCorrupted);\n }\n let mut bytes = [0; HEADER_SIZE];\n self.reader.read_exact(&mut bytes)?;\n self.header = Header::parse(&bytes)?;\n self.header.validate()?;\n self.hash.update(&bytes);\n if !self.header.no_checksum() {\n self.rolling_checksum = CHECKSUM_FLAG;\n }\n self.state = DecoderState::Pages;\n Ok(())\n }\n\n pub(crate) fn decode_page(&mut self, data: &mut [u8]) -> Result> {\n if self.state == DecoderState::Close {\n return Ok(None);\n }\n if self.state != DecoderState::Pages || data.len() != self.header.page_size as usize {\n return Err(Error::LTXCorrupted);\n }\n\n let mut header_bytes = [0; PAGE_HEADER_SIZE];\n self.reader.read_exact(&mut header_bytes)?;\n let page = PageHeader::parse(&header_bytes)?;\n self.hash.update(&header_bytes);\n if page.is_zero() {\n self.state = DecoderState::Close;\n return Ok(None);\n }\n page.validate()?;\n\n if page.flags & PAGE_HEADER_FLAG_SIZE != 0 {\n let mut size_bytes = [0; 4];\n self.reader.read_exact(&mut size_bytes)?;\n self.hash.update(&size_bytes);\n let compressed_size = u32::from_be_bytes(size_bytes) as usize;\n if compressed_size > crate::lz4_block::compress_bound(data.len()) {\n return Err(Error::LTXCorrupted);\n }\n let mut compressed = vec![0; compressed_size];\n self.reader.read_exact(&mut compressed)?;\n let n = lz4_flex::block::decompress_into(&compressed, data)\n .map_err(|_| Error::LTXCorrupted)?;\n if n != data.len() {\n return Err(Error::LTXCorrupted);\n }\n } else {\n let mut decoder = lz4_flex::frame::FrameDecoder::new(&mut self.reader);\n decoder.read_exact(data).map_err(|_| Error::LTXCorrupted)?;\n let mut extra = [0; 1];\n if decoder.read(&mut extra).map_err(|_| Error::LTXCorrupted)? != 0 {\n return Err(Error::LTXCorrupted);\n }\n }\n\n self.hash.update(data);\n if self.header.is_snapshot()\n && !self.header.no_checksum()\n && page.pgno != lock_pgno(self.header.page_size)\n {\n self.rolling_checksum =\n CHECKSUM_FLAG | (self.rolling_checksum ^ checksum_page(page.pgno, data));\n }\n Ok(Some(page))\n }\n\n pub(crate) fn close(&mut self) -> Result<()> {\n if self.state == DecoderState::Closed {\n return Ok(());\n }\n if self.state != DecoderState::Close {\n return Err(Error::LTXCorrupted);\n }\n\n let mut remaining = Vec::new();\n self.reader.read_to_end(&mut remaining)?;\n if remaining.len() < 8 + TRAILER_SIZE {\n return Err(Error::LTXCorrupted);\n }\n\n let trailer_offset = remaining.len() - TRAILER_SIZE;\n let size_offset = trailer_offset - 8;\n let index_size = u64::from_be_bytes(\n remaining[size_offset..trailer_offset]\n .try_into()\n .map_err(|_| Error::LTXCorrupted)?,\n ) as usize;\n if index_size != size_offset {\n return Err(Error::LTXCorrupted);\n }\n verify_page_index(&remaining[..size_offset])?;\n\n self.trailer = Trailer::parse(&remaining[trailer_offset..])?;\n self.trailer.validate(self.header)?;\n self.hash\n .update(&remaining[..remaining.len() - CHECKSUM_SIZE]);\n if CHECKSUM_FLAG | self.hash.sum64() != self.trailer.file_checksum {\n return Err(Error::ChecksumMismatch);\n }\n if self.header.is_snapshot()\n && !self.header.no_checksum()\n && self.rolling_checksum != self.trailer.post_apply_checksum\n {\n return Err(Error::ChecksumMismatch);\n }\n\n self.state = DecoderState::Closed;\n Ok(())\n }\n}\n\npub(crate) struct Encoder {\n pub(crate) writer: W,\n pub(crate) header: Header,\n pub(crate) trailer: Trailer,\n hash: Crc64,\n index: BTreeMap,\n compressor: crate::lz4_block::Compressor,\n page_encoding: PageEncoding,\n bytes_written: u64,\n previous_page_number: u32,\n header_written: bool,\n closed: bool,\n}\n\nimpl Encoder {\n pub(crate) fn new_legacy(writer: W) -> Self {\n Self::new(writer, PageEncoding::LegacyFrame)\n }\n\n pub(crate) fn new_block(writer: W) -> Self {\n Self::new(writer, PageEncoding::SizedBlock)\n }\n\n fn new(writer: W, page_encoding: PageEncoding) -> Self {\n Self {\n writer,\n header: Header::default(),\n trailer: Trailer::default(),\n hash: Crc64::new(),\n index: BTreeMap::new(),\n compressor: crate::lz4_block::Compressor::default(),\n page_encoding,\n bytes_written: 0,\n previous_page_number: 0,\n header_written: false,\n closed: false,\n }\n }\n\n pub(crate) fn encode_header(&mut self, header: Header) -> Result<()> {\n if self.header_written || self.closed {\n return Err(Error::LTXCorrupted);\n }\n header.validate()?;\n self.header = header;\n let bytes = header.marshal();\n self.write_hashed(&bytes)?;\n self.header_written = true;\n Ok(())\n }\n\n pub(crate) fn encode_page(&mut self, mut page: PageHeader, data: &[u8]) -> Result<()> {\n if !self.header_written\n || self.closed\n || page.pgno > self.header.commit\n || data.len() != self.header.page_size as usize\n {\n return Err(Error::LTXCorrupted);\n }\n page.validate()?;\n let lock_page = lock_pgno(self.header.page_size);\n if page.pgno == lock_page {\n return Err(Error::LTXCorrupted);\n }\n\n if self.header.is_snapshot() {\n if self.previous_page_number == 0 && page.pgno != 1 {\n return Err(Error::LTXCorrupted);\n }\n let expected = if self.previous_page_number == lock_page - 1 {\n self.previous_page_number + 2\n } else {\n self.previous_page_number + 1\n };\n if self.previous_page_number != 0 && page.pgno != expected {\n return Err(Error::LTXCorrupted);\n }\n } else if self.previous_page_number >= page.pgno {\n return Err(Error::LTXCorrupted);\n }\n\n let offset = self.bytes_written;\n let compressed = match self.page_encoding {\n PageEncoding::LegacyFrame => {\n if page.flags != 0 {\n return Err(Error::LTXCorrupted);\n }\n self.write_hashed(&page.marshal())?;\n // The pre-v0.5.2 Go encoder uses independent 64 KiB blocks and\n // writes a content checksum. Litestream's legacy decoder expects\n // the resulting eight-byte frame trailer (end mark + checksum).\n let frame_info = lz4_flex::frame::FrameInfo::new()\n .block_size(lz4_flex::frame::BlockSize::Max64KB)\n .block_mode(lz4_flex::frame::BlockMode::Independent)\n .content_checksum(true);\n let mut encoder =\n lz4_flex::frame::FrameEncoder::with_frame_info(frame_info, Vec::new());\n encoder.write_all(data)?;\n encoder.finish().map_err(|_| Error::LTXCorrupted)?\n }\n PageEncoding::SizedBlock => {\n let compressed = self.compressor.compress(data);\n page.flags |= PAGE_HEADER_FLAG_SIZE;\n self.write_hashed(&page.marshal())?;\n let size = u32::try_from(compressed.len())\n .map_err(|_| Error::LTXCorrupted)?\n .to_be_bytes();\n self.write_hashed(&size)?;\n compressed\n }\n };\n self.writer.write_all(&compressed)?;\n self.bytes_written += compressed.len() as u64;\n self.hash.update(data);\n\n self.previous_page_number = page.pgno;\n self.index\n .insert(page.pgno, (offset, self.bytes_written - offset));\n Ok(())\n }\n\n pub(crate) fn close(&mut self, post_apply_checksum: u64) -> Result<()> {\n if !self.header_written || self.closed {\n return Err(Error::LTXCorrupted);\n }\n\n self.write_hashed(&[0; PAGE_HEADER_SIZE])?;\n let index_offset = self.bytes_written;\n let mut index_bytes = Vec::new();\n for (&page_number, &(offset, size)) in &self.index {\n write_uvarint(&mut index_bytes, page_number as u64);\n write_uvarint(&mut index_bytes, offset);\n write_uvarint(&mut index_bytes, size);\n }\n write_uvarint(&mut index_bytes, 0);\n self.write_hashed(&index_bytes)?;\n self.write_hashed(&(self.bytes_written - index_offset).to_be_bytes())?;\n\n self.trailer.post_apply_checksum = post_apply_checksum;\n self.hash.update(&post_apply_checksum.to_be_bytes());\n self.trailer.file_checksum = CHECKSUM_FLAG | self.hash.sum64();\n self.trailer.validate(self.header)?;\n if self.header.commit == 0 && post_apply_checksum != CHECKSUM_FLAG {\n return Err(Error::LTXCorrupted);\n }\n self.writer.write_all(&self.trailer.marshal())?;\n self.bytes_written += TRAILER_SIZE as u64;\n self.closed = true;\n Ok(())\n }\n\n fn write_hashed(&mut self, bytes: &[u8]) -> Result<()> {\n self.writer.write_all(bytes)?;\n self.hash.update(bytes);\n self.bytes_written += bytes.len() as u64;\n Ok(())\n }\n}\n\nfn verify_page_index(bytes: &[u8]) -> Result<()> {\n decode_page_index(bytes).map(|_| ())\n}\n\n/// Decodes a page index into its `(pgno, frame_offset, frame_size)` entries.\n/// The index is varint triples terminated by a `pgno == 0` marker; each entry\n/// locates one page's whole frame (header plus LZ4 body) at `frame_offset` in\n/// the file, `frame_size` bytes long. Paged restore builds its page map from\n/// these, so it can range-read one frame without the object.\npub(crate) fn decode_page_index(bytes: &[u8]) -> Result> {\n let mut position = 0;\n let mut entries = Vec::new();\n loop {\n let page_number = read_uvarint(bytes, &mut position)?;\n if page_number == 0 {\n break;\n }\n let offset = read_uvarint(bytes, &mut position)?;\n let size = read_uvarint(bytes, &mut position)?;\n entries.push((page_number as u32, offset, size));\n }\n if position != bytes.len() {\n return Err(Error::LTXCorrupted);\n }\n Ok(entries)\n}\n\n/// Decodes one page frame — the `[header][optional 4-byte size][LZ4 body]` a\n/// page-index entry locates — into `page_size` bytes, standalone from the\n/// sequential [`Decoder`]. It reads the same two page encodings the decoder\n/// does (a sized LZ4 block or a legacy LZ4 frame), so paged restore serves a\n/// single ranged-read frame without materializing the object.\n/// Decodes the frame that a page index says holds `pgno`. The frame names\n/// its own page, so a wrong index entry or offset is an error, not another\n/// page's bytes served as this one.\npub(crate) fn decode_page_frame_of(frame: &[u8], page_size: usize, pgno: u32) -> Result> {\n let header = frame.get(..PAGE_HEADER_SIZE).ok_or(Error::LTXCorrupted)?;\n if PageHeader::parse(header)?.pgno != pgno {\n return Err(Error::LTXCorrupted);\n }\n decode_page_frame(frame, page_size)\n}\n\npub(crate) fn decode_page_frame(frame: &[u8], page_size: usize) -> Result> {\n let mut reader = std::io::Cursor::new(frame);\n let mut header_bytes = [0; PAGE_HEADER_SIZE];\n reader.read_exact(&mut header_bytes)?;\n let page = PageHeader::parse(&header_bytes)?;\n if page.is_zero() {\n return Err(Error::LTXCorrupted);\n }\n page.validate()?;\n\n let mut data = vec![0u8; page_size];\n if page.flags & PAGE_HEADER_FLAG_SIZE != 0 {\n let mut size_bytes = [0; 4];\n reader.read_exact(&mut size_bytes)?;\n let compressed_size = u32::from_be_bytes(size_bytes) as usize;\n if compressed_size > crate::lz4_block::compress_bound(page_size) {\n return Err(Error::LTXCorrupted);\n }\n let mut compressed = vec![0; compressed_size];\n reader.read_exact(&mut compressed)?;\n let n = lz4_flex::block::decompress_into(&compressed, &mut data)\n .map_err(|_| Error::LTXCorrupted)?;\n if n != page_size {\n return Err(Error::LTXCorrupted);\n }\n } else {\n let mut decoder = lz4_flex::frame::FrameDecoder::new(&mut reader);\n decoder\n .read_exact(&mut data)\n .map_err(|_| Error::LTXCorrupted)?;\n }\n Ok(data)\n}\n\nfn read_uvarint(bytes: &[u8], position: &mut usize) -> Result {\n let mut value = 0u64;\n let mut shift = 0u32;\n loop {\n let byte = *bytes.get(*position).ok_or(Error::LTXCorrupted)?;\n *position += 1;\n if byte < 0x80 {\n if shift >= 64 || (shift == 63 && byte > 1) {\n return Err(Error::LTXCorrupted);\n }\n return Ok(value | (u64::from(byte) << shift));\n }\n value |= u64::from(byte & 0x7f) << shift;\n shift += 7;\n if shift >= 70 {\n return Err(Error::LTXCorrupted);\n }\n }\n}\n\nfn write_uvarint(bytes: &mut Vec, mut value: u64) {\n while value >= 0x80 {\n bytes.push(value as u8 | 0x80);\n value >>= 7;\n }\n bytes.push(value as u8);\n}\n","editable",false,"branch","documentNavigation","repo","denoland/celld","repoId","a23b8a10-92cf-4dc4-a7ab-a04e58298dde","ownerId","org:42048915","site","nexthub","private","admin","settings","maintainer","triage","starred","stars",1,"watchers","watching","pinned","notify","views",[134,135,136,137,138,139,140,141,142,143,144,145,146],"navigation",{"_55":56,"_91":92,"_93":94,"_95":96,"_97":98,"_99":100,"_101":102},"sources",[],"viewer","meta",{"_83":84,"_85":86,"_87":88,"_43":32,"_89":14,"_90":32,"_49":50,"_51":50},"branches",[14],"tags",[70,71,72,73,74,75,76,77,78,79,80,81,82],"cloneUrl","https://git.nexthub.sh/denoland/celld.git","v0.0.1","v0.0.2","v0.1.0","v0.2.0","v0.2.1","v0.3.0","v0.4.0","v0.4.1","v0.5.0","v0.5.1","v0.6.0","v0.6.1","v0.6.2","description","self-hosted, distributed Durable Objects","owner","denoland","section","","head","empty","pins",[116,117,118],"documents",{"_111":112,"_113":114},"code",{"_108":109,"_110":104},"automations",{"_106":50,"_107":-5},"delivery",{"_103":104,"_105":-5},"tag","7gTA52Z9a1vO6FP6eCQPix9zO4O","releases",0,"applications","workflows","deployments","commits",13,"changes","counts",{"_115":114},"total",38,"issue",{"_119":110,"_121":132,"_15":110,"_123":133,"_53":125,"_126":32,"_127":125,"_128":104},{"_119":129,"_121":130,"_15":129,"_123":131,"_53":125,"_126":32,"_127":125,"_128":104},{"_119":120,"_121":122,"_15":120,"_123":124,"_53":125,"_126":32,"_127":125,"_128":114},"id","issues","name","Issues","filters",{},true,"saved","pinnable","count","tasks","Tasks",{},"Changes",{},{"_119":110,"_121":132,"_15":110,"_123":177,"_53":125,"_126":32,"_127":125},{"_119":129,"_121":130,"_15":129,"_123":176,"_53":125,"_126":32,"_127":125},{"_119":120,"_121":122,"_15":120,"_123":175,"_53":125,"_126":32,"_127":125},{"_119":172,"_121":173,"_15":172,"_123":174,"_53":32,"_126":32,"_127":125},{"_119":169,"_121":170,"_15":169,"_123":171,"_53":32,"_126":32,"_127":125},{"_119":106,"_121":167,"_15":97,"_123":168,"_53":32,"_126":32,"_127":125},{"_119":107,"_121":164,"_15":165,"_123":166,"_53":32,"_126":32,"_127":125},{"_119":103,"_121":162,"_15":103,"_123":163,"_53":32,"_126":32,"_127":125},{"_119":158,"_121":159,"_15":160,"_123":161,"_53":32,"_126":32,"_127":125},{"_119":93,"_121":156,"_15":93,"_123":157,"_53":32,"_126":32,"_127":32},{"_119":153,"_121":154,"_15":153,"_123":155,"_53":32,"_126":32,"_127":32},{"_119":150,"_121":151,"_15":150,"_123":152,"_53":32,"_126":32,"_127":32},{"_119":147,"_121":148,"_15":147,"_123":149,"_53":32,"_126":32,"_127":32},"decisions","Decisions",{},"changelog","Changelog",{},"roadmap","Roadmap",{},"Documents",{},"activity","Activity","insights",{},"Releases",{},"Deployments","automations/deployments",{},"Workflows",{},"discussions","Discussions",{},"goals","Goals",{},{},{},{},"groups","user","agents",[],"now",1794435658467,"sidebar","expanded"]