indexbus_inspect/internal/

journal_dump.rs

use core::mem::size_of;
use std::fs::File;
use std::path::Path;
use std::time::Duration;

use indexbus_abi::layouts::JournalLayout4;
use indexbus_abi::LayoutHeader;
use memmap2::Mmap;

use crate::Error;

const HDR_BYTES: usize = 8;
const FLAG_COMMITTED: u32 = 1;
const FLAG_PAD: u32 = 2;

#[inline]
fn align8(n: u32) -> u32 {
    (n + 7) & !7
}

#[inline]
fn record_bytes(payload_len_bytes: u32) -> u32 {
    align8((HDR_BYTES as u32) + payload_len_bytes)
}

#[inline]
fn read_u32(p: *const u8) -> u32 {
    unsafe { core::ptr::read_unaligned(p as *const u32) }
}

fn read_header(base: *const u8, offset: usize) -> (u32, u32) {
    let payload_len_bytes = read_u32(unsafe { base.add(offset) });
    let flags = read_u32(unsafe { base.add(offset + 4) });
    (payload_len_bytes, flags)
}

fn read_layout_header(m: &Mmap) -> Result<LayoutHeader, Error> {
    if m.len() < size_of::<LayoutHeader>() {
        return Err(Error::TooSmall {
            needed: size_of::<LayoutHeader>(),
            got: m.len(),
        });
    }
    let ptr = m.as_ptr() as *const LayoutHeader;
    let header = unsafe { core::ptr::read_unaligned(ptr) };
    if !header.is_compatible_v1() {
        return Err(Error::IncompatibleLayout);
    }
    Ok(header)
}

pub(crate) fn dump_journal_path(
    path: &Path,
    start_pos: u64,
    max_records: usize,
) -> Result<(), Error> {
    let file = File::open(path)?;
    let mapped = unsafe { Mmap::map(&file)? };

    let header = read_layout_header(&mapped)?;
    if (header.capabilities & indexbus_abi::caps::INDEXBUS_CAP_SUPPORTS_JOURNAL) == 0 {
        return Err(Error::UnsupportedLayout {
            capabilities: header.capabilities,
            layout_bytes: header.layout_bytes,
        });
    }

    if mapped.len() < size_of::<JournalLayout4>() {
        return Err(Error::TooSmall {
            needed: size_of::<JournalLayout4>(),
            got: mapped.len(),
        });
    }

    let layout = mapped.as_ptr() as *const JournalLayout4;
    let segment_bytes = unsafe { (*layout).segment_bufs[0].len() as u32 };

    println!("journal dump: start_pos={start_pos} max_records={max_records}");
    println!(
        "segment_bytes={segment_bytes} segments={} subs={}",
        indexbus_abi::INDEXBUS_JOURNAL_SEGMENTS,
        indexbus_abi::INDEXBUS_JOURNAL_SUBSCRIBERS_DEFAULT
    );

    let mut pos = start_pos;
    for n in 0..max_records {
        let off = (pos as u32) % segment_bytes;
        let segment_idx =
            ((pos / segment_bytes as u64) as usize) % indexbus_abi::INDEXBUS_JOURNAL_SEGMENTS;

        let segment_base = unsafe { (*layout).segment_bufs[segment_idx].as_ptr() };
        let (payload_len, flags) = read_header(segment_base, off as usize);

        if payload_len == 0 {
            println!("{n}: pos={pos} segment={segment_idx} off={off} <empty>");
            break;
        }

        let rbytes = record_bytes(payload_len);
        let committed = (flags & FLAG_COMMITTED) != 0;
        let pad = (flags & FLAG_PAD) != 0;

        println!(
            "{n}: pos={pos} segment={segment_idx} off={off} payload_len={payload_len} record_bytes={rbytes} flags=0x{flags:08x} committed={committed} pad={pad}"
        );

        if !committed {
            break;
        }

        pos += rbytes as u64;
    }

    Ok(())
}

pub(crate) fn watch_journal_path(path: &Path, interval: Duration) -> Result<(), Error> {
    let file = File::open(path)?;
    let mapped = unsafe { Mmap::map(&file)? };

    let header = read_layout_header(&mapped)?;
    if (header.capabilities & indexbus_abi::caps::INDEXBUS_CAP_SUPPORTS_JOURNAL) == 0 {
        return Err(Error::UnsupportedLayout {
            capabilities: header.capabilities,
            layout_bytes: header.layout_bytes,
        });
    }

    if mapped.len() < size_of::<JournalLayout4>() {
        return Err(Error::TooSmall {
            needed: size_of::<JournalLayout4>(),
            got: mapped.len(),
        });
    }

    let layout = mapped.as_ptr() as *const JournalLayout4;
    let base = mapped.as_ptr();

    let stats_present =
        (header.capabilities & indexbus_abi::caps::INDEXBUS_CAP_SUPPORTS_JOURNAL_STATS) != 0;

    // Compute stats offset using the same ordering rule as validation: base -> optional wake -> stats.
    let mut stats_offset = size_of::<JournalLayout4>().div_ceil(64) * 64;
    if (header.capabilities & indexbus_abi::caps::INDEXBUS_CAP_SUPPORTS_BLOCKING) != 0 {
        stats_offset = (stats_offset + size_of::<indexbus_abi::layouts::JournalWakeSection4>())
            .div_ceil(64)
            * 64;
    }

    let mut last_pub_pos: u64 =
        unsafe { core::ptr::read_unaligned(core::ptr::addr_of!((*layout).pub_pos) as *const u64) };
    let mut last_records: u64 = 0;

    println!("watching journal (interval={:?})", interval);
    println!(
        "caps=0x{:08x} stats_present={stats_present}",
        header.capabilities
    );

    loop {
        std::thread::sleep(interval);

        let pub_pos: u64 = unsafe {
            core::ptr::read_unaligned(core::ptr::addr_of!((*layout).pub_pos) as *const u64)
        };
        let delta_bytes = pub_pos.saturating_sub(last_pub_pos);
        last_pub_pos = pub_pos;

        let bytes_per_sec = (delta_bytes as f64) / interval.as_secs_f64();

        if stats_present {
            // Best-effort: only if the mapping is big enough.
            let needed = stats_offset + size_of::<indexbus_abi::layouts::JournalStatsSection4>();
            if mapped.len() >= needed {
                let off = stats_offset;
                let payload_bytes = unsafe {
                    core::ptr::read_unaligned(base.add(
                        off + core::mem::offset_of!(
                            indexbus_abi::layouts::JournalStatsSection4,
                            pub_payload_bytes
                        ),
                    ) as *const u64)
                };
                let records = unsafe {
                    core::ptr::read_unaligned(base.add(
                        off + core::mem::offset_of!(
                            indexbus_abi::layouts::JournalStatsSection4,
                            pub_records
                        ),
                    ) as *const u64)
                };

                let d_records = records.saturating_sub(last_records);
                last_records = records;

                let recs_per_sec = (d_records as f64) / interval.as_secs_f64();

                println!(
                    "pub_pos={pub_pos} bytes/sec={bytes_per_sec:.0} records/sec={recs_per_sec:.0} total_payload_bytes={payload_bytes} total_records={records}",
                );
                continue;
            }
        }

        println!("pub_pos={pub_pos} bytes/sec={:.0}", bytes_per_sec);
    }
}