Read the next record from the mmap'd region. Returns `None` at EOF or at the first corruption point. Zero-copy: payload bytes reference the mmap'd region directly.
(&mut self)
| 145 | /// Returns `None` at EOF or at the first corruption point. |
| 146 | /// Zero-copy: payload bytes reference the mmap'd region directly. |
| 147 | pub fn next_record(&mut self) -> Result<Option<WalRecord>> { |
| 148 | let data = &self.mmap[..]; |
| 149 | |
| 150 | loop { |
| 151 | // Check if we have enough bytes for a header. |
| 152 | if self.offset + HEADER_SIZE > data.len() { |
| 153 | return Ok(None); |
| 154 | } |
| 155 | |
| 156 | // Parse header. |
| 157 | let header_bytes: &[u8; HEADER_SIZE] = data[self.offset..self.offset + HEADER_SIZE] |
| 158 | .try_into() |
| 159 | .map_err(|_| { |
| 160 | WalError::Io(std::io::Error::new( |
| 161 | std::io::ErrorKind::InvalidData, |
| 162 | "header slice conversion failed", |
| 163 | )) |
| 164 | })?; |
| 165 | let header = RecordHeader::from_bytes(header_bytes); |
| 166 | |
| 167 | // Validate magic — corruption or end of valid data. |
| 168 | if header.magic != WAL_MAGIC { |
| 169 | return Ok(None); |
| 170 | } |
| 171 | |
| 172 | // Validate version. |
| 173 | if header.validate(self.offset as u64).is_err() { |
| 174 | return Ok(None); |
| 175 | } |
| 176 | |
| 177 | let payload_len = header.payload_len as usize; |
| 178 | let record_end = self.offset + HEADER_SIZE + payload_len; |
| 179 | |
| 180 | // Check if payload is fully within the mmap'd region. |
| 181 | if record_end > data.len() { |
| 182 | return Ok(None); // Torn write at segment end. |
| 183 | } |
| 184 | |
| 185 | // Extract payload (copies from mmap to owned Vec). |
| 186 | let payload = data[self.offset + HEADER_SIZE..record_end].to_vec(); |
| 187 | self.offset = record_end; |
| 188 | |
| 189 | let record = WalRecord { header, payload }; |
| 190 | |
| 191 | // Verify checksum. |
| 192 | if record.verify_checksum().is_err() { |
| 193 | return Ok(None); // Corruption — end of committed prefix. |
| 194 | } |
| 195 | |
| 196 | // Check record type. |
| 197 | let logical_type = record.logical_record_type(); |
| 198 | if RecordType::from_raw(logical_type).is_none() { |
| 199 | if RecordType::is_required(logical_type) { |
| 200 | return Err(WalError::UnknownRequiredRecordType { |
| 201 | record_type: header.record_type, |
| 202 | lsn: header.lsn, |
| 203 | }); |
| 204 | } |
no test coverage detected