p2p skeleton and starting to put together stuff for

Author: Anirudh Vyas

src/core/meta.rs

@@ -33,7 +33,7 @@ pub struct SegmentMeta {
 unsafe impl Zeroable for SegmentMeta {}
 unsafe impl Pod for SegmentMeta {}
 
-/// Cursor tracking a subscriber’s read position in the segment log.
+/// Cursor tracking a subscriber's read position in the segment log.
 ///
 /// Fields:
 /// - `subscriber_name`: fixed-size identifier of the subscriber.

src/core/mod.rs

@@ -43,7 +43,7 @@ pub trait XaeroData: Any + Send + Sync + Clone + Debug {}
 impl<T> XaeroData for T where T: Any + Send + Sync + Clone + Debug {}
 
 /// Global, singleton configuration instance.
-/// 
+///
 /// Initialized by `load_config` and reused thereafter.
 pub static CONF: OnceLock<config::Config> = OnceLock::new();
 
@@ -84,7 +84,7 @@ pub fn init_global_io_pool() {
 /// - Loads and validates configuration (`xaeroflux.toml`).
 /// - Initializes dispatcher and I/O thread pools.
 /// - Sets up logging and displays startup banner.
-/// 
+///
 /// # Panics
 /// Will panic if the configuration name is not "xaeroflux".
 pub fn initialize() {

src/indexing/storage/actors/mmr_actor.rs

@@ -19,6 +19,7 @@ use crate::{
         hash::sha_256,
         storage::{format::archive, mmr::XaeroMmrOps},
     },
+    system::CONTROL_BUS,
 };
 
 /// Actor responsible for indexing events into a Merkle Mountain Range (MMR).
@@ -38,7 +39,7 @@ pub struct MmrIndexingActor {
 impl MmrIndexingActor {
     /// Create a new `MmrIndexingActor` with optional store and listener.
     ///
-    /// If `store` is `None`, a default `SegmentWriterActor` with prefix "mmr" is used.
+    /// If `store` is `None`, a default `SegmentWriterActor` with prefix "xaeroflux-mmr" is used.
     /// If `listener` is `None`, an `EventListener` is created that:
     ///   1. Archives each event into bytes.
     ///   2. Persists the event into LMDB.
@@ -54,7 +55,7 @@ impl MmrIndexingActor {
         let _mmr = Arc::new(Mutex::new(crate::indexing::storage::mmr::XaeroMmr::new()));
         let _store = Arc::new(store.unwrap_or_else(|| {
             SegmentWriterActor::new_with_config(super::segment_writer_actor::SegmentConfig {
-                prefix: "mmr".to_string(),
+                prefix: "xaeroflux-mmr".to_string(),
                 ..Default::default()
             })
         }));

src/indexing/storage/actors/mod.rs

@@ -2,3 +2,4 @@ pub mod indexing_actor;
 pub mod mmr_actor;
 pub mod segment_reader_actor;
 pub mod segment_writer_actor;
+pub mod secondary_index_actor;

src/indexing/storage/actors/secondary_index_actor.rs

@@ -0,0 +1,53 @@
+use std::{
+    collections::HashMap,
+    sync::{Arc, Mutex},
+    time::{Duration, Instant},
+};
+
+use crate::{
+    core::{aof::LmdbEnv, event::Event, listeners::EventListener, meta::SegmentMeta, IO_POOL},
+    system::control_bus::SystemPayload,
+};
+
+pub struct SecondaryIndexActor {
+    /// cache for the secondary index.
+    pub(crate) cache: HashMap<[u8; 32], (Option<SegmentMeta>, bool, Instant)>,
+    pub(crate) lmdb_env: Arc<Mutex<LmdbEnv>>,
+    // The TTL for the garbage collection of the secondary index.
+    pub(crate) gc_ttl: Duration,
+    pub listener: EventListener<SystemPayload>,
+}
+
+impl SecondaryIndexActor {
+    pub fn new(lmdb_env: Arc<Mutex<LmdbEnv>>, gc_ttl: Duration) -> Self {
+        let (tx, rx) = crossbeam::channel::unbounded::<Event<SystemPayload>>();
+        let txc = tx.clone();
+        let listener = EventListener::new(
+            "secondary_index_actor",
+            Arc::new(move |e: Event<SystemPayload>| {
+                txc.send(e).expect("failed to send event");
+            }),
+            None,
+            Some(1), // single-threaded handler
+        );
+
+        IO_POOL.get()
+        .expect("IO_POOL not initialized")
+        .execute(move || {
+            let env = lmdb_env.lock().expect("Failed to lock LMDB environment");
+            env.dbis[1]
+        });
+
+        Self {
+            cache: HashMap::new(),
+            lmdb_env,
+            gc_ttl,
+            listener,
+        }
+    }
+
+    /// Checks if the secondary index is empty.
+    pub fn is_empty(&self) -> bool {
+        self.cache.is_empty()
+    }
+}

src/indexing/storage/actors/segment_writer_actor.rs

@@ -2,10 +2,10 @@
 //!
 //! This module provides:
 //! - `SegmentConfig`: configuration parameters for paging and segmentation.
-//! - `SegmentWriterActor`: an actor that listens for archived event blobs
-//!   and writes them into fixed-size pages within segment files.
-//! - `run_writer_loop`: core loop logic for handling page boundaries,
-//!   file rollover, and metadata events.
+//! - `SegmentWriterActor`: an actor that listens for archived event blobs and writes them into
+//!   fixed-size pages within segment files.
+//! - `run_writer_loop`: core loop logic for handling page boundaries, file rollover, and metadata
+//!   events.
 //! - Unit tests verifying segment math, flush behavior, and rollover/resume logic.
 
 use std::{
@@ -28,6 +28,7 @@ use crate::{
         size::PAGE_SIZE,
     },
     indexing::storage::format::archive,
+    system::{CONTROL_BUS, control_bus::SystemPayload},
 };
 
 /// Configuration for paged segment storage.
@@ -64,8 +65,8 @@ impl Default for SegmentConfig {
 /// Actor responsible for writing serialized event frames into segment files.
 ///
 /// This actor sets up:
-/// - An `EventListener<Vec<u8>>` that archives incoming `Event<Vec<u8>>` values
-///   and sends their byte frames to `inbox`.
+/// - An `EventListener<Vec<u8>>` that archives incoming `Event<Vec<u8>>` values and sends their
+///   byte frames to `inbox`.
 /// - A background thread (`_handle`) running `run_writer_loop` to consume from `inbox`.
 /// - An LMDB environment (`meta_db`) to record segment rollover metadata.
 /// - The active `SegmentConfig` used for page size and directory settings.
@@ -153,9 +154,9 @@ impl SegmentWriterActor {
 /// 2. Iterates existing `SegmentMeta` entries to resume at the latest segment/page.
 /// 3. Opens (or creates) the current segment file, memory-maps it.
 /// 4. On each `rx.recv()`:
-///    - If the incoming frame does not fit in the current page, flushes the page,
-///      advances `page_index`, and handles segment rollover (including
-///      emitting a `MetaEvent` into LMDB and remapping a new file).
+///    - If the incoming frame does not fit in the current page, flushes the page, advances
+///      `page_index`, and handles segment rollover (including emitting a `MetaEvent` into LMDB and
+///      remapping a new file).
 ///    - Copies the frame bytes into the current page at `byte_offset + write_pos`.
 ///    - Updates `write_pos`.
 /// 5. On channel close, performs a final flush of any unwritten bytes.
@@ -267,8 +268,37 @@ fn run_writer_loop(meta_db: &Arc<Mutex<LmdbEnv>>, rx: Receiver<Vec<u8>>, config:
         let end = start + write_len;
         mm[start..end].copy_from_slice(data.as_slice());
         write_pos += write_len;
+        tracing::debug!(
+            "Wrote {} bytes to segment {} page {} at offset {}",
+            write_len,
+            seg_id,
+            local_page_idx,
+            start
+        );
+        tracing::debug!("sending message to control bus");
+        let payload_written_msg_sent_ack = CONTROL_BUS
+            .get()
+            .expect("control_bus_not_initialized")
+            .sender()
+            .send(SystemPayload::PayloadWritten {
+                meta: SegmentMeta {
+                    page_index,
+                    segment_index: seg_id,
+                    write_pos,
+                    byte_offset,
+                    latest_segment_id: seg_id,
+                    ts_start,
+                    ts_end: emit_secs(),
+                },
+            });
+        match payload_written_msg_sent_ack {
+            Ok(_) => tracing::debug!("Payload written message sent successfully"),
+            Err(e) => {
+                // FIXME: SHOULD WE PANIC HERE??
+                tracing::error!("Failed to send PayloadWritten message: {}", e)
+            }
+        }
     }
-
     // final flush if any data
     if write_pos > 0 {
         mm.flush_range(byte_offset, page_size)

src/indexing/storage/format.rs

@@ -162,7 +162,7 @@ pub fn unarchive<'a>(bytes: &'a [u8]) -> (&'a XaeroOnDiskEventHeader, &'a Archiv
 /// For each leaf (event) we store exactly where its bytes live on disk
 /// and what its timestamp was.
 #[repr(C)]
-#[derive(Clone, Copy)]
+#[derive(Debug, Clone, Copy)]
 pub struct LeafLocation {
     /// Which payload segment file ('.seg') holds the event
     pub segment_index: u32, // 4 bytes

src/indexing/storage/mmr.rs

@@ -1,8 +1,11 @@
 use rkyv::Archive;
 
-use crate::indexing::{
-    hash::sha_256_concat_hash,
-    merkle_tree::{XaeroMerkleProof, XaeroMerkleTree, XaeroMerkleTreeOps},
+use crate::{
+    indexing::{
+        hash::sha_256_concat_hash,
+        merkle_tree::{XaeroMerkleProof, XaeroMerkleTree, XaeroMerkleTreeOps},
+    },
+    system::{CONTROL_BUS, control_bus::SystemPayload},
 };
 
 /// Peak represents a peak in the Merkle Mountain Range (MMR).
@@ -64,6 +67,9 @@ pub trait XaeroMmrOps {
         proof: &XaeroMerkleProof,
         expected_root: [u8; 32],
     ) -> bool;
+
+    /// Convenience: gets the leaf hash by its index.
+    fn get_leaf_hash_by_index(&self, index: usize) -> Option<&[u8; 32]>;
 }
 
 impl Default for XaeroMmr {
@@ -119,6 +125,22 @@ impl XaeroMmrOps for XaeroMmr {
         // bag peaks and recalculate root
         let t = XaeroMerkleTree::neo(self.peaks.iter().map(|p| p.root).collect());
         self.root = t.root();
+        let res = self.leaf_hashes.iter().enumerate().map(|(i, h)| {
+            tracing::debug!("leaf_hashes[{}]: {:?}", i, h);
+            CONTROL_BUS
+                .get()
+                .expect("CONTROL_BUS must be initialized")
+                .sender()
+                .send(SystemPayload::MmrAppended {
+                    leaf_index: i as u64,
+                    leaf_hash: *h,
+                })
+        });
+        res.for_each(|r| {
+            if let Err(e) = r {
+                tracing::error!("Failed to send MmrAppended event: {:?}", e);
+            }
+        });
         changed
     }
 
@@ -203,6 +225,10 @@ impl XaeroMmrOps for XaeroMmr {
         }
         running == expected_root
     }
+
+    fn get_leaf_hash_by_index(&self, index: usize) -> Option<&[u8; 32]> {
+        self.leaf_hashes.get(index)
+    }
 }
 
 #[cfg(test)]

src/lib.rs

@@ -3,7 +3,7 @@ pub mod indexing;
 pub mod logs;
 pub mod networking;
 pub mod sys;
-
+pub mod system;
 use core::{DISPATCHER_POOL, aof::AOFActor, event::Event};
 use std::{
     sync::{
@@ -23,6 +23,7 @@ use indexing::storage::{
     },
     format::archive,
 };
+use system::{CONTROL_BUS, init_control_bus};
 use threadpool::ThreadPool;
 
 static NEXT_ID: AtomicU64 = AtomicU64::new(1);
@@ -149,6 +150,10 @@ impl Subject {
     ///
     /// Returns an `XFluxHandle` that keeps everything alive until dropped.
     pub fn unsafe_run(self: Arc<Self>) -> XFluxHandle {
+        tracing::debug!("unsafe_run called for Subject: {}", self.name);
+        tracing::debug!("initializing control bus");
+        init_control_bus();
+        tracing::debug!("control bus initialized");
         self.unsafe_run_called.store(true, Ordering::SeqCst);
         // Instantiate system actors
         let aof = Arc::new(AOFActor::new());

src/system/control_bus.rs

@@ -1,19 +1,91 @@
+use std::sync::Arc;
+
 use bytemuck::{Pod, Zeroable};
-use crate::core::event::{LeafLocation, SystemErrorCode};
+use crossbeam::channel::{Receiver, Sender, unbounded};
+
+use crate::core::{DISPATCHER_POOL, meta::SegmentMeta};
 
 /// The data carried by each system event.
 /// All fields are fixed-size so this enum is Pod-friendly.
 #[repr(C)]
-#[derive(Debug, Clone, Copy, Archive, Serialize, Deserialize, Pod, Zeroable)]
-#[archive_bounds(::core::fmt::Debug)]
+#[derive(Debug, Clone, Copy)]
 pub enum SystemPayload {
-    PayloadWritten { leaf_index: u64, loc: LeafLocation },
-    SegmentRolledOver { ts_start: u64, seg_idx: u32 },
-    SegmentRollOverFailed { ts_start: u64, seg_idx: u32, error_code: u16 },
-    PageFlushed { page_index: u32 },
-    PageFlushFailed { page_index: u32, error_code: u16 },
-    MmrAppended { leaf_index: u64, leaf_hash: [u8;32] },
-    MmrAppendFailed { leaf_index: u64, error_code: u16 },
-    SecondaryIndexWritten { leaf_hash: [u8;32] },
-    SecondaryIndexFailed { leaf_hash: [u8;32], error_code: u16 },
-}
+    /// Indicates an event payload was written to a segment page.
+    PayloadWritten {
+        meta: SegmentMeta,
+    },
+    SegmentRolledOver {
+        meta: SegmentMeta,
+    },
+    SegmentRollOverFailed {
+        meta: SegmentMeta,
+        error_code: u16,
+    },
+    PageFlushed {
+        meta: SegmentMeta,
+    },
+    PageFlushFailed {
+        meta: SegmentMeta,
+        error_code: u16,
+    },
+    MmrAppended {
+        leaf_index: u64,
+        leaf_hash: [u8; 32],
+    },
+    MmrAppendFailed {
+        leaf_index: u64,
+        error_code: u16,
+    },
+    SecondaryIndexWritten {
+        leaf_hash: [u8; 32],
+    },
+    SecondaryIndexFailed {
+        leaf_hash: [u8; 32],
+        error_code: u16,
+    },
+}
+
+unsafe impl Zeroable for SystemPayload {}
+unsafe impl Pod for SystemPayload {}
+
+pub struct ControlBus {
+    pub(crate) tx: Sender<SystemPayload>,
+    pub(crate) rx: Receiver<SystemPayload>,
+}
+
+impl ControlBus {
+    /// Create a new ControlBus with an unbounded channel.
+    pub fn new() -> Self {
+        let (tx, rx) = unbounded();
+        Self { tx, rx }
+    }
+
+    /// Get a sender handle for publishing system payloads.
+    pub fn sender(&self) -> Sender<SystemPayload> {
+        self.tx.clone()
+    }
+
+    /// Subscribe to receive all system payloads.
+    pub fn subscribe(&self) -> Receiver<SystemPayload> {
+        self.rx.clone()
+    }
+
+    /// Register a handler that will be called for every system payload.
+    pub fn register_handler<H: ControlHandler + Send + Sync + 'static>(&self, handler: Arc<H>) {
+        let rx = self.rx.clone();
+        DISPATCHER_POOL
+            .get()
+            .expect("DISPATCH POOL not initialized")
+            .execute(move || {
+                while let Ok(payload) = rx.recv() {
+                    handler.handle(payload);
+                }
+            });
+    }
+}
+
+/// Trait that actors implement to handle incoming system payloads.
+pub trait ControlHandler {
+    /// Called for each system payload received on the control bus.
+    fn handle(&self, payload: SystemPayload);
+}