byteor_pipeline_backings_shm/

journal.rs

//! SHM-backed journal backings (v1, optional).
//!
//! This module is feature-gated because journals are optional in the v1 OSS lane-graph.

use std::path::Path;

use byteor_pipeline_kernel::{LaneRx, LaneTx, LaneTxError};

use crate::{AttachOptions, BackingError};

/// SHM-backed journal publisher.
///
/// Attach-only: opens/creates the mapping and exposes a publisher handle.
pub struct ShmJournalPublisher {
    _region: indexbus_transport_shm::JournalRegion4,
    pubr: indexbus_log::JournalPublisher,
}

impl ShmJournalPublisher {
    /// Append a payload (claim + copy + commit).
    #[inline]
    pub fn try_append(&mut self, payload: &[u8]) -> Result<u64, BackingError> {
        Ok(self.pubr.try_append(payload)?)
    }

    /// Current publication position.
    #[inline]
    pub fn pub_pos(&self) -> u64 {
        self.pubr.pub_pos()
    }

    /// Maximum payload size supported by this layout.
    #[inline]
    pub fn max_payload(&self) -> u32 {
        self.pubr.max_payload()
    }
}

impl LaneTx for ShmJournalPublisher {
    fn publish(&mut self, msg: &[u8]) -> Result<(), LaneTxError> {
        let max = self.max_payload() as usize;
        if msg.len() > max {
            return Err(LaneTxError::TooLarge {
                max,
                len: msg.len(),
            });
        }

        self.try_append(msg).map(|_| ()).map_err(|e| match e {
            BackingError::Journal(indexbus_log::Error::Full) => LaneTxError::Full,
            _ => LaneTxError::Failed,
        })
    }
}

/// SHM-backed journal subscriber.
///
/// Attach-only: opens/creates the mapping and exposes a subscriber handle.
pub struct ShmJournalSubscriber {
    _region: indexbus_transport_shm::JournalRegion4,
    sub: indexbus_log::JournalSubscriber,
}

impl ShmJournalSubscriber {
    /// Poll for the next record, copying bytes into `out`.
    ///
    /// Returns `Ok(None)` when no record is available yet.
    #[inline]
    pub fn poll_next_len(&mut self, out: &mut [u8]) -> Result<Option<usize>, BackingError> {
        Ok(self.sub.poll_next_len(out)?)
    }

    /// Spin/poll until the next record is available.
    #[inline]
    pub fn wait_next_len<W: indexbus_core::WaitStrategy>(
        &mut self,
        out: &mut [u8],
        wait: &mut W,
    ) -> Result<usize, BackingError> {
        let slice = self.sub.wait_next(out, wait)?;
        Ok(slice.len())
    }

    /// Current subscriber position.
    #[inline]
    pub fn pos(&self) -> u64 {
        self.sub.pos()
    }

    /// Subscriber lag behind the publisher.
    #[inline]
    pub fn lag_bytes(&self) -> u64 {
        self.sub.lag_bytes()
    }
}

impl LaneRx for ShmJournalSubscriber {
    fn recv(&mut self, buf: &mut [u8]) -> Option<usize> {
        match self.poll_next_len(buf) {
            Ok(Some(n)) => Some(n),
            Ok(None) => None,
            Err(_) => None,
        }
    }
}

fn transport_open_options(opts: &AttachOptions) -> indexbus_transport_shm::JournalOpenOptions {
    let mut o = indexbus_transport_shm::JournalOpenOptions::new().blocking(opts.blocking);
    if let Some(p) = &opts.path {
        o = o.path(p.clone());
    }
    o
}

fn resolve_path<'a>(opts: &'a AttachOptions, lane: &'a str) -> Option<&'a Path> {
    let _ = lane;
    opts.path.as_deref()
}

/// Attach to a journal publisher.
pub fn attach_journal_publisher(
    opts: &AttachOptions,
    lane: &str,
    cfg: indexbus_log::JournalPublisherConfig,
) -> Result<ShmJournalPublisher, BackingError> {
    if lane.is_empty() {
        return Err(BackingError::Invalid("lane name is empty"));
    }

    let o = transport_open_options(opts);
    let mut region = if let Some(path) = resolve_path(opts, lane) {
        indexbus_transport_shm::JournalRegion4::open_path_with(path, o)?
    } else {
        indexbus_transport_shm::JournalRegion4::open_with(lane, o)?
    };

    let pubr = region.publisher(cfg)?;
    Ok(ShmJournalPublisher {
        _region: region,
        pubr,
    })
}

/// Attach to a journal subscriber starting at an explicit position.
pub fn attach_journal_subscriber_from(
    opts: &AttachOptions,
    lane: &str,
    start_pos: u64,
) -> Result<ShmJournalSubscriber, BackingError> {
    if lane.is_empty() {
        return Err(BackingError::Invalid("lane name is empty"));
    }

    let o = transport_open_options(opts);
    let mut region = if let Some(path) = resolve_path(opts, lane) {
        indexbus_transport_shm::JournalRegion4::open_path_with(path, o)?
    } else {
        indexbus_transport_shm::JournalRegion4::open_with(lane, o)?
    };

    let sub = region.subscriber(opts.queue, start_pos)?;
    Ok(ShmJournalSubscriber {
        _region: region,
        sub,
    })
}

/// Attach to a journal subscriber starting at the current tail (safe-by-default).
///
/// This avoids unintentionally replaying old data unless an operator explicitly requests it
/// via [`attach_journal_subscriber_from`].
pub fn attach_journal_subscriber_tail(
    opts: &AttachOptions,
    lane: &str,
) -> Result<ShmJournalSubscriber, BackingError> {
    if lane.is_empty() {
        return Err(BackingError::Invalid("lane name is empty"));
    }

    let o = transport_open_options(opts);
    let mut region = if let Some(path) = resolve_path(opts, lane) {
        indexbus_transport_shm::JournalRegion4::open_path_with(path, o)?
    } else {
        indexbus_transport_shm::JournalRegion4::open_with(lane, o)?
    };

    // Determine current tail position without publishing.
    let tail = {
        let pubr = region.publisher(indexbus_log::JournalPublisherConfig {
            subs: 1,
            ..Default::default()
        })?;
        pubr.pub_pos()
    };

    let sub = region.subscriber(opts.queue, tail)?;
    Ok(ShmJournalSubscriber {
        _region: region,
        sub,
    })
}