Method next_record

nodedb-wal/src/reader.rs:83–147

Read the next record from the WAL. Returns `None` at EOF (clean end) or at the first corruption point. Returns `Err` only for I/O errors or unknown required record types.

(&mut self)

    /// Returns `None` at EOF (clean end) or at the first corruption point.
    /// Returns `Err` only for I/O errors or unknown required record types.
    pub fn next_record(&mut self) -> Result<Option<WalRecord>> {
        loop {
            // Read header.
            let mut header_buf = [0u8; HEADER_SIZE];
            match self.read_exact(&mut header_buf) {
                Ok(()) => {}
                Err(WalError::Io(e)) if e.kind() == std::io::ErrorKind::UnexpectedEof => {
                    return Ok(None); // Clean EOF.
                }
                Err(e) => return Err(e),
            }

            let header = RecordHeader::from_bytes(&header_buf);

            // Validate magic and version.
            if header.validate(self.offset - HEADER_SIZE as u64).is_err() {
                // Corruption or end of valid data — treat as end of committed prefix.
                return Ok(None);
            }

            // Read payload.
            let mut payload = vec![0u8; header.payload_len as usize];
            if !payload.is_empty() {
                match self.read_exact(&mut payload) {
                    Ok(()) => {}
                    Err(WalError::Io(e)) if e.kind() == std::io::ErrorKind::UnexpectedEof => {
                        return Ok(None);
                    }
                    Err(e) => return Err(e),
                }
            }

            let record = WalRecord { header, payload };

            // Verify checksum.
            if record.verify_checksum().is_err() {
                if let Some(dwb) = &mut self.double_write
                    && let Ok(Some(recovered)) = dwb.recover_record(header.lsn)
                {
                    tracing::info!(
                        lsn = header.lsn,
                        "recovered torn write from double-write buffer"
                    );
                    self.offset += recovered.payload.len() as u64;
                    return Ok(Some(recovered));
                }
                return Ok(None);
            }

            // Check if the record type is known (strip encrypted flag for lookup).
            let logical_type = record.logical_record_type();
            if RecordType::from_raw(logical_type).is_none() {
                if RecordType::is_required(logical_type) {
                    return Err(WalError::UnknownRequiredRecordType {
                        record_type: header.record_type,
                        lsn: header.lsn,
                    });
                }