byteor_pipeline_backings_mem/

lib.rs

#![deny(missing_docs)]
#![deny(unreachable_pub, rust_2018_idioms)]
#![forbid(unsafe_code)]

//! In-memory lane backings for deterministic worker-side flows.
//!
//! These backings avoid file-backed SHM entirely and are intended for Plane B style workflows
//! such as compile/validate/replay helpers where OS-level shared memory is unnecessary.

use std::collections::VecDeque;
use std::sync::{Arc, Mutex};

use byteor_pipeline_kernel::{
    LaneRx, LaneRxBorrow, LaneRxError, LaneTx, LaneTxError, LaneTxWith, LaneTxWithError,
    LaneTxWithResult,
};

/// Options for an in-memory events lane.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct MemLaneOptions {
    /// Maximum queued messages before publishes return `LaneTxError::Full`.
    pub capacity: usize,
    /// Maximum message length accepted by the lane.
    pub max_message_len: usize,
}

impl Default for MemLaneOptions {
    fn default() -> Self {
        Self {
            capacity: 1024,
            max_message_len: indexbus_core::INDEXBUS_SLOT_DATA_SIZE,
        }
    }
}

#[derive(Default)]
struct QueueState {
    msgs: VecDeque<Vec<u8>>,
}

/// In-memory events TX lane.
#[derive(Clone)]
pub struct MemEventsTx {
    state: Arc<Mutex<QueueState>>,
    options: MemLaneOptions,
}

/// In-memory events RX lane.
#[derive(Clone)]
pub struct MemEventsRx {
    state: Arc<Mutex<QueueState>>,
}

/// Create an in-memory events channel with default message sizing.
pub fn events_channel(capacity: usize) -> (MemEventsTx, MemEventsRx) {
    events_channel_with_options(MemLaneOptions {
        capacity,
        ..MemLaneOptions::default()
    })
}

/// Create an in-memory events channel with explicit options.
pub fn events_channel_with_options(options: MemLaneOptions) -> (MemEventsTx, MemEventsRx) {
    let state = Arc::new(Mutex::new(QueueState::default()));
    (
        MemEventsTx {
            state: Arc::clone(&state),
            options,
        },
        MemEventsRx { state },
    )
}

impl MemEventsRx {
    /// Return the current queued depth.
    pub fn len(&self) -> usize {
        self.state.lock().expect("mem lane poisoned").msgs.len()
    }

    /// Whether the lane is empty.
    pub fn is_empty(&self) -> bool {
        self.len() == 0
    }
}

impl LaneRx for MemEventsRx {
    fn recv(&mut self, buf: &mut [u8]) -> Option<usize> {
        let msg = self
            .state
            .lock()
            .expect("mem lane poisoned")
            .msgs
            .pop_front()?;
        let n = msg.len().min(buf.len());
        buf[..n].copy_from_slice(&msg[..n]);
        Some(n)
    }
}

impl LaneRxBorrow for MemEventsRx {
    fn recv_with<F, R>(&mut self, f: F) -> Result<Option<R>, LaneRxError>
    where
        F: FnOnce(&[u8]) -> R,
    {
        let Some(msg) = self
            .state
            .lock()
            .expect("mem lane poisoned")
            .msgs
            .pop_front()
        else {
            return Ok(None);
        };
        Ok(Some(f(&msg)))
    }
}

impl LaneTx for MemEventsTx {
    fn publish(&mut self, msg: &[u8]) -> Result<(), LaneTxError> {
        if msg.len() > self.options.max_message_len {
            return Err(LaneTxError::TooLarge {
                max: self.options.max_message_len,
                len: msg.len(),
            });
        }

        let mut state = self.state.lock().expect("mem lane poisoned");
        if state.msgs.len() >= self.options.capacity {
            return Err(LaneTxError::Full);
        }
        state.msgs.push_back(msg.to_vec());
        Ok(())
    }
}

impl LaneTxWith for MemEventsTx {
    fn publish_with<F>(&mut self, f: F) -> Result<(), LaneTxError>
    where
        F: FnOnce(&mut [u8]) -> usize,
    {
        let mut slot = vec![0u8; self.options.max_message_len];
        let n = f(&mut slot);
        if n > self.options.max_message_len {
            return Err(LaneTxError::TooLarge {
                max: self.options.max_message_len,
                len: n,
            });
        }
        self.publish(&slot[..n])
    }
}

impl LaneTxWithResult for MemEventsTx {
    fn publish_with_result<F, E>(&mut self, f: F) -> Result<(), LaneTxWithError<E>>
    where
        F: FnOnce(&mut [u8]) -> Result<usize, E>,
    {
        let mut slot = vec![0u8; self.options.max_message_len];
        let n = f(&mut slot).map_err(LaneTxWithError::Encode)?;
        if n > self.options.max_message_len {
            return Err(LaneTxWithError::Failed);
        }
        self.publish(&slot[..n]).map_err(Into::into)
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn mem_events_channel_round_trips_fifo() {
        let (mut tx, mut rx) = events_channel(4);
        tx.publish(b"a").unwrap();
        tx.publish(b"b").unwrap();

        let mut buf = [0u8; 8];
        let n = rx.recv(&mut buf).unwrap();
        assert_eq!(&buf[..n], b"a");
        let n = rx.recv(&mut buf).unwrap();
        assert_eq!(&buf[..n], b"b");
        assert!(rx.recv(&mut buf).is_none());
    }

    #[test]
    fn mem_events_channel_reports_backpressure() {
        let (mut tx, rx) = events_channel(1);
        tx.publish(b"a").unwrap();
        assert_eq!(tx.publish(b"b"), Err(LaneTxError::Full));
        assert_eq!(rx.len(), 1);
    }

    #[test]
    fn mem_events_channel_supports_borrow_and_slot_writes() {
        let (mut tx, mut rx) = events_channel_with_options(MemLaneOptions {
            capacity: 4,
            max_message_len: 16,
        });

        tx.publish_with(|slot| {
            slot[..3].copy_from_slice(b"abc");
            3
        })
        .unwrap();

        let n = rx.recv_with(|msg| msg.len()).unwrap().unwrap();
        assert_eq!(n, 3);
    }

    #[test]
    fn mem_events_channel_supports_fallible_slot_writes() {
        let (mut tx, mut rx) = events_channel(4);
        tx.publish_with_result(|slot| {
            slot[..2].copy_from_slice(b"ok");
            Ok::<usize, &'static str>(2)
        })
        .unwrap();

        let msg = rx.recv_with(|msg| msg.to_vec()).unwrap().unwrap();
        assert_eq!(msg, b"ok".to_vec());

        let err = tx
            .publish_with_result(|_slot| Err::<usize, _>("encode failed"))
            .unwrap_err();
        assert_eq!(err, LaneTxWithError::Encode("encode failed"));
    }
}