Function read_log_segment_from_bytes

nodedb/src/engine/timeseries/reader.rs:234–291

Read and decode a log segment from in-memory bytes.

(
    data: &[u8],
    registry: &DictionaryRegistry,
)

/// Read and decode a log segment from in-memory bytes.
pub fn read_log_segment_from_bytes(
    data: &[u8],
    registry: &DictionaryRegistry,
) -> Result<LogSegmentData, SegmentReadError> {
    let header_end = decode_tseg_header(data)?;
    let mut cur = Cursor::new(&data[header_end..]);

    let kind = cur
        .read_u8()
        .ok_or(SegmentReadError::TooSmall { size: data.len() })?;
    if kind != KIND_LOG {
        return Err(SegmentReadError::UnknownKind { kind });
    }

    let entry_count = cur
        .read_u32_le()
        .ok_or(SegmentReadError::TooSmall { size: data.len() })? as usize;
    let compressed_len = cur
        .read_u32_le()
        .ok_or(SegmentReadError::TooSmall { size: data.len() })? as usize;

    let compressed_block = cur
        .read_bytes(compressed_len)
        .ok_or(SegmentReadError::TooSmall { size: data.len() })?;

    let raw = decompress_log(compressed_block, registry).map_err(|e| {
        SegmentReadError::Decompression {
            detail: e.to_string(),
        }
    })?;

    // Parse raw bytes back into log entries.
    // Format per entry: [timestamp_ms:8] [data_len:4] [data:N]
    let mut entries = Vec::with_capacity(entry_count);
    let mut raw_cur = Cursor::new(&raw);

    while raw_cur.remaining() >= 12 && entries.len() < entry_count {
        let timestamp_ms = raw_cur.read_u64_le().expect(
            "invariant: 'while raw_cur.remaining() >= 12' loop guard ensures 8 bytes for timestamp",
        ) as i64;
        let data_len = raw_cur
            .read_u32_le()
            .expect("invariant: remaining >= 12, u64 consumed 8, leaving 4 for data_len")
            as usize;

        let entry_data = match raw_cur.read_bytes(data_len) {
            Some(b) => b.to_vec(),
            None => break,
        };

        entries.push(LogEntry {
            timestamp_ms,
            data: entry_data,
        });
    }

    Ok(LogSegmentData { entries })
}