indexbus_inspect/internal/

inspect.rs

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

use indexbus_abi::layouts::{
    FanoutWakeSection4, JournalLayout4, JournalStatsSection4, JournalWakeSection4, SegmentMeta,
    SequencerLayout4, SequencerWakeSection4, SharedFanoutLayout4, SharedLayout, SharedWakeSection,
    StateLayout256, StateWakeSection,
};
use indexbus_abi::LayoutHeader;
use indexbus_abi::{caps, flags};
use memmap2::Mmap;

use crate::Error;

#[derive(Debug, Clone)]
pub(crate) struct WakeInfo {
    pub offset: usize,
    pub size: usize,
    pub present: bool,
}

#[derive(Debug, Clone)]
pub(crate) struct Report {
    pub path: String,
    pub kind: &'static str,

    pub header_magic: u32,
    pub header_version: u16,
    pub header_flags: u16,

    pub capabilities: u32,
    pub cap_names: Vec<&'static str>,

    pub layout_bytes: u32,
    pub mapped_bytes: usize,

    pub base_size: usize,
    pub initialized: Option<u32>,
    pub wake: Option<WakeInfo>,

    pub offsets: Vec<(&'static str, usize)>,

    /// Kind-specific sampled values (best-effort).
    pub values: Vec<(&'static str, String)>,
}

#[inline]
const fn align_up(value: usize, align: usize) -> usize {
    debug_assert!(align.is_power_of_two());
    (value + (align - 1)) & !(align - 1)
}

fn cap_names(c: u32) -> Vec<&'static str> {
    let mut out = Vec::new();
    if (c & caps::INDEXBUS_CAP_SUPPORTS_EVENTS) != 0 {
        out.push("EVENTS");
    }
    if (c & caps::INDEXBUS_CAP_SUPPORTS_STATE) != 0 {
        out.push("STATE");
    }
    if (c & caps::INDEXBUS_CAP_SUPPORTS_FANOUT) != 0 {
        out.push("FANOUT");
    }
    if (c & caps::INDEXBUS_CAP_SUPPORTS_BLOCKING) != 0 {
        out.push("BLOCKING");
    }
    if (c & caps::INDEXBUS_CAP_SUPPORTS_SEQUENCER) != 0 {
        out.push("SEQUENCER");
    }
    if (c & caps::INDEXBUS_CAP_SUPPORTS_JOURNAL) != 0 {
        out.push("JOURNAL");
    }
    if (c & caps::INDEXBUS_CAP_SUPPORTS_JOURNAL_STATS) != 0 {
        out.push("JOURNAL_STATS");
    }
    out
}

fn read_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)
}

fn classify_kind(header: &LayoutHeader, mapped_bytes: usize) -> Option<(&'static str, usize)> {
    let caps_bits = header.capabilities;
    let lb = header.layout_bytes as usize;
    let effective = mapped_bytes.min(lb.max(size_of::<LayoutHeader>()));

    let kind = flags::region_kind(header.flags);
    match kind {
        flags::INDEXBUS_REGION_KIND_JOURNAL => {
            let base = size_of::<JournalLayout4>();
            (effective >= base).then_some(("journal4", base))
        }
        flags::INDEXBUS_REGION_KIND_SEQUENCER => {
            let base = size_of::<SequencerLayout4>();
            (effective >= base).then_some(("sequencer4", base))
        }
        flags::INDEXBUS_REGION_KIND_FANOUT => {
            let base = size_of::<SharedFanoutLayout4>();
            (effective >= base).then_some(("fanout4", base))
        }
        flags::INDEXBUS_REGION_KIND_EVENTS => {
            let base = size_of::<SharedLayout>();
            (effective >= base).then_some(("events", base))
        }
        flags::INDEXBUS_REGION_KIND_STATE => {
            let base = size_of::<StateLayout256>();
            if effective >= base {
                Some(("state256", base))
            } else {
                // State layouts are const-parameterized in Rust; we only recognize the concrete 256B one.
                Some(("state(unknown)", base))
            }
        }
        _ => {
            // Backward-compat fallback: older mappings may not set the region-kind discriminator
            // in `flags` yet. Preserve the previous heuristic classification so golden fixtures
            // and legacy files remain inspectable.

            if (caps_bits & caps::INDEXBUS_CAP_SUPPORTS_JOURNAL) != 0 {
                let base = size_of::<JournalLayout4>();
                if effective >= base {
                    return Some(("journal4", base));
                }
            }

            if (caps_bits & caps::INDEXBUS_CAP_SUPPORTS_SEQUENCER) != 0 {
                let base = size_of::<SequencerLayout4>();
                if effective >= base {
                    return Some(("sequencer4", base));
                }
            }

            if (caps_bits & caps::INDEXBUS_CAP_SUPPORTS_FANOUT) != 0 {
                let base = size_of::<SharedFanoutLayout4>();
                if (caps_bits & caps::INDEXBUS_CAP_SUPPORTS_EVENTS) != 0 && effective >= base {
                    return Some(("fanout4", base));
                }
            }

            if (caps_bits & caps::INDEXBUS_CAP_SUPPORTS_EVENTS) != 0 {
                let base = size_of::<SharedLayout>();
                if effective >= base {
                    return Some(("events", base));
                }
            }

            if (caps_bits & caps::INDEXBUS_CAP_SUPPORTS_STATE) != 0 {
                let base = size_of::<StateLayout256>();
                if effective >= base {
                    return Some(("state256", base));
                }
                return Some(("state(unknown)", base));
            }

            None
        }
    }
}

fn read_u32_at(base: *const u8, offset: usize) -> u32 {
    // Safe as a raw read: used for inspection only.
    unsafe { core::ptr::read_unaligned(base.add(offset) as *const u32) }
}

fn read_u64_at(base: *const u8, offset: usize) -> u64 {
    unsafe { core::ptr::read_unaligned(base.add(offset) as *const u64) }
}

fn values_for_kind(kind: &str, base: *const u8) -> Vec<(&'static str, String)> {
    let mut out = Vec::new();

    match kind {
        "journal4" => {
            let pub_pos = read_u64_at(base, core::mem::offset_of!(JournalLayout4, pub_pos));
            out.push(("pub_pos", pub_pos.to_string()));

            let active_segment =
                read_u32_at(base, core::mem::offset_of!(JournalLayout4, active_segment));
            out.push(("active_segment", active_segment.to_string()));

            // Sample segment metadata (best-effort).
            let mut segment_ids = Vec::new();
            let mut segment_tails = Vec::new();
            for i in 0..3 {
                let seg_base = core::mem::offset_of!(JournalLayout4, segments)
                    + i * core::mem::size_of::<SegmentMeta>();
                let id = read_u32_at(
                    base,
                    seg_base + core::mem::offset_of!(SegmentMeta, segment_id),
                );
                let tail = read_u64_at(base, seg_base + core::mem::offset_of!(SegmentMeta, tail));
                segment_ids.push(id);
                segment_tails.push(tail);
            }
            out.push(("segment_id", format!("{:?}", segment_ids)));
            out.push(("segment_tail", format!("{:?}", segment_tails)));

            // Sample subscriber positions (best-effort).
            let mut sub_pos = Vec::new();
            for i in 0..4 {
                let off = core::mem::offset_of!(JournalLayout4, sub_pos)
                    + i * core::mem::size_of::<u64>();
                sub_pos.push(read_u64_at(base, off));
            }
            out.push(("sub_pos", format!("{:?}", sub_pos)));

            if let Some(min) = sub_pos.iter().copied().min() {
                out.push(("sub_pos_min", min.to_string()));
            }
            if let Some(max) = sub_pos.iter().copied().max() {
                out.push(("sub_pos_max", max.to_string()));
            }
        }
        "sequencer4" => {
            let cursor = read_u64_at(base, core::mem::offset_of!(SequencerLayout4, cursor));
            out.push(("cursor", cursor.to_string()));

            // Best-effort: sample all gating sequences.
            let mut gating_vals = Vec::new();
            for i in 0..4 {
                let off = core::mem::offset_of!(SequencerLayout4, gating)
                    + i * core::mem::size_of::<indexbus_abi::layouts::SequencerGatingCell>();
                let v = read_u64_at(base, off);
                gating_vals.push(v);
            }
            out.push(("gating", format!("{:?}", gating_vals)));

            if let Some(min) = gating_vals.iter().copied().min() {
                out.push(("gating_min", min.to_string()));
            }
        }
        _ => {}
    }

    out
}

fn read_initialized(kind: &str, base: *const u8) -> Option<u32> {
    match kind {
        "events" => Some(read_u32_at(
            base,
            core::mem::offset_of!(SharedLayout, initialized),
        )),
        "fanout4" => Some(read_u32_at(
            base,
            core::mem::offset_of!(SharedFanoutLayout4, initialized),
        )),
        "sequencer4" => Some(read_u32_at(
            base,
            core::mem::offset_of!(SequencerLayout4, initialized),
        )),
        "journal4" => Some(read_u32_at(
            base,
            core::mem::offset_of!(JournalLayout4, initialized),
        )),
        _ => None,
    }
}

fn wake_info(
    kind: &str,
    header: &LayoutHeader,
    mapped_bytes: usize,
    base_size: usize,
) -> Option<WakeInfo> {
    if (header.capabilities & caps::INDEXBUS_CAP_SUPPORTS_BLOCKING) == 0 {
        return None;
    }

    let wake_size = match kind {
        "events" => size_of::<SharedWakeSection>(),
        "fanout4" => size_of::<FanoutWakeSection4>(),
        "state256" | "state(unknown)" => size_of::<StateWakeSection>(),
        "sequencer4" => size_of::<SequencerWakeSection4>(),
        "journal4" => size_of::<JournalWakeSection4>(),
        _ => return None,
    };

    let wake_offset = align_up(base_size, 64);

    let claimed = header.layout_bytes as usize;
    let effective = mapped_bytes.min(claimed.max(size_of::<LayoutHeader>()));

    let present = effective >= wake_offset.saturating_add(wake_size);

    Some(WakeInfo {
        offset: wake_offset,
        size: wake_size,
        present,
    })
}

fn offsets_for_kind(kind: &str) -> Vec<(&'static str, usize)> {
    let mut out = Vec::new();

    match kind {
        "events" => {
            out.push((
                "SharedLayout.slot_pool",
                core::mem::offset_of!(SharedLayout, slot_pool),
            ));
            out.push((
                "SharedLayout.queue",
                core::mem::offset_of!(SharedLayout, queue),
            ));
            out.push((
                "SharedLayout.mpsc_queue",
                core::mem::offset_of!(SharedLayout, mpsc_queue),
            ));
        }
        "fanout4" => {
            out.push((
                "SharedFanoutLayout4.slot_pool",
                core::mem::offset_of!(SharedFanoutLayout4, slot_pool),
            ));
            out.push((
                "SharedFanoutLayout4.producer_queue",
                core::mem::offset_of!(SharedFanoutLayout4, producer_queue),
            ));
            out.push((
                "SharedFanoutLayout4.consumer_queues",
                core::mem::offset_of!(SharedFanoutLayout4, consumer_queues),
            ));
        }
        "sequencer4" => {
            out.push((
                "SequencerLayout4.cursor",
                core::mem::offset_of!(SequencerLayout4, cursor),
            ));
            out.push((
                "SequencerLayout4.gating",
                core::mem::offset_of!(SequencerLayout4, gating),
            ));
        }
        "journal4" => {
            out.push((
                "JournalLayout4.pub_pos",
                core::mem::offset_of!(JournalLayout4, pub_pos),
            ));
            out.push((
                "JournalLayout4.active_segment",
                core::mem::offset_of!(JournalLayout4, active_segment),
            ));
            out.push((
                "JournalLayout4.segments",
                core::mem::offset_of!(JournalLayout4, segments),
            ));
            out.push((
                "JournalLayout4.sub_pos",
                core::mem::offset_of!(JournalLayout4, sub_pos),
            ));
            out.push((
                "JournalLayout4.segment_bufs",
                core::mem::offset_of!(JournalLayout4, segment_bufs),
            ));
            out.push(("SegmentMeta.tail", core::mem::offset_of!(SegmentMeta, tail)));
        }
        "state256" => {
            out.push((
                "StateLayout256.seq",
                core::mem::offset_of!(StateLayout256, seq),
            ));
            out.push((
                "StateLayout256.len",
                core::mem::offset_of!(StateLayout256, len),
            ));
            out.push((
                "StateLayout256.data",
                core::mem::offset_of!(StateLayout256, data),
            ));
        }
        _ => {}
    }

    out
}

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

    let header = read_header(&mapped)?;

    let (kind, base_size) =
        classify_kind(&header, mapped.len()).ok_or(Error::UnsupportedLayout {
            capabilities: header.capabilities,
            layout_bytes: header.layout_bytes,
        })?;

    let base = mapped.as_ptr();
    let init = read_initialized(kind, base);

    let wake = wake_info(kind, &header, mapped.len(), base_size);

    let offsets = offsets_for_kind(kind);
    let mut values = values_for_kind(kind, base);

    // Best-effort: if a journal stats section is appended, sample its counters.
    if kind == "journal4" && (header.capabilities & caps::INDEXBUS_CAP_SUPPORTS_JOURNAL_STATS) != 0
    {
        let stats_size = size_of::<JournalStatsSection4>();
        let mut stats_offset = align_up(base_size, 64);
        if (header.capabilities & caps::INDEXBUS_CAP_SUPPORTS_BLOCKING) != 0 {
            stats_offset = align_up(stats_offset + size_of::<JournalWakeSection4>(), 64);
        }

        let claimed = header.layout_bytes as usize;
        let effective = mapped.len().min(claimed.max(size_of::<LayoutHeader>()));
        let present = effective >= stats_offset.saturating_add(stats_size);

        if present {
            let b = mapped.as_ptr();
            let off = stats_offset;

            let pub_payload_bytes = read_u64_at(
                b,
                off + core::mem::offset_of!(JournalStatsSection4, pub_payload_bytes),
            );
            let pub_records = read_u64_at(
                b,
                off + core::mem::offset_of!(JournalStatsSection4, pub_records),
            );
            let pad_records = read_u64_at(
                b,
                off + core::mem::offset_of!(JournalStatsSection4, pad_records),
            );
            let rotations = read_u64_at(
                b,
                off + core::mem::offset_of!(JournalStatsSection4, rotations),
            );
            let pub_would_overrun = read_u64_at(
                b,
                off + core::mem::offset_of!(JournalStatsSection4, pub_would_overrun),
            );
            let pub_forced_catchup = read_u64_at(
                b,
                off + core::mem::offset_of!(JournalStatsSection4, pub_forced_catchup),
            );

            values.push(("stats.offset", stats_offset.to_string()));
            values.push(("stats.pub_payload_bytes", pub_payload_bytes.to_string()));
            values.push(("stats.pub_records", pub_records.to_string()));
            values.push(("stats.pad_records", pad_records.to_string()));
            values.push(("stats.rotations", rotations.to_string()));
            values.push(("stats.pub_would_overrun", pub_would_overrun.to_string()));
            values.push(("stats.pub_forced_catchup", pub_forced_catchup.to_string()));

            let mut sub_overruns = Vec::new();
            for i in 0..4 {
                let o = core::mem::offset_of!(JournalStatsSection4, sub_overruns)
                    + i * core::mem::size_of::<u64>();
                sub_overruns.push(read_u64_at(b, off + o));
            }
            values.push(("stats.sub_overruns", format!("{:?}", sub_overruns)));
        }
    }

    Ok(Report {
        path: path.to_string_lossy().into_owned(),
        kind,

        header_magic: header.magic,
        header_version: header.version,
        header_flags: header.flags,

        capabilities: header.capabilities,
        cap_names: cap_names(header.capabilities),

        layout_bytes: header.layout_bytes,
        mapped_bytes: mapped.len(),

        base_size,
        initialized: init,
        wake,

        offsets,
        values,
    })
}