sdk_api_read ready - the replay / scan is wired using segment reader actor as a 'first' operation that is is run on materializer materialize so that events are sinked before any events start flowing

Author: Anirudh Vyas

src/core/aof.rs

@@ -3,6 +3,34 @@ pub const MS_PER_HOUR: u64 = 3_600_000; // 3 600 s * 1 000 ms
 pub const MS_PER_MINUTE: u64 = 60_000; // 60 s * 1 000 ms
 pub const MS_PER_SECOND: u64 = 1_000; // 1 s * 1 000 ms
 
+/// Shortcut for epoch ms.
+pub fn emit_secs() -> u64 {
+    std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .expect("failed to get timestamp")
+        .as_secs()
+}
+
+pub fn emit_millis() -> u128 {
+    std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .expect("failed to get timestamp")
+        .as_millis()
+}
+
+pub fn emit_micros() -> u128 {
+    std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .expect("failed to get timestamp")
+        .as_micros()
+}
+
+pub fn emit_nanos() -> u128 {
+    std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .expect("failed to get timestamp")
+        .as_nanos()
+}
 /// Given any epoch‐ms, return (start_of_utc_day, start_of_next_utc_day).
 /// All in UTC.  If you need local‐midnight instead, just add/subtract your fixed offset.
 pub fn day_bounds_from_epoch_ms(ms: u64) -> (u64, u64) {

src/core/date_time.rs

@@ -32,6 +32,7 @@ pub enum SystemEventKind {
     Resume,
     Shutdown,
     Restart,
+    Replay, // Replay is a special event that is used to replay the event log
 }
 
 impl EventType {
@@ -48,8 +49,9 @@ impl EventType {
             4 => EventType::SystemEvent(SystemEventKind::Resume),
             5 => EventType::SystemEvent(SystemEventKind::Shutdown),
             6 => EventType::SystemEvent(SystemEventKind::Restart),
-            7 => EventType::NetworkEvent(value),
-            8 => EventType::StorageEvent(value),
+            7 => EventType::SystemEvent(SystemEventKind::Replay),
+            8 => EventType::NetworkEvent(value),
+            9 => EventType::StorageEvent(value),
             _ => panic!("Invalid event type"),
         }
     }
@@ -63,6 +65,7 @@ impl EventType {
             EventType::SystemEvent(SystemEventKind::Resume) => 4,
             EventType::SystemEvent(SystemEventKind::Shutdown) => 5,
             EventType::SystemEvent(SystemEventKind::Restart) => 6,
+            EventType::SystemEvent(SystemEventKind::Replay) => 7,
             EventType::NetworkEvent(v) => *v,
             EventType::StorageEvent(v) => *v,
             EventType::MetaEvent(v) => META_BASE + *v,

src/core/event.rs

@@ -1,16 +1,16 @@
 use bytemuck::{Pod, Zeroable};
 use rkyv::{Archive, Deserialize, Serialize};
 
-#[repr(C)]
-#[derive(Debug, Clone, Archive, Serialize, Deserialize, Default)]
-#[rkyv(derive(Debug))]
-#[derive(Copy)]
+#[repr(C, packed)]
+#[derive(Debug, Clone, Default, Copy)]
 pub struct SegmentMeta {
     pub page_index: usize,
     pub segment_index: usize,
     pub write_pos: usize,
     pub byte_offset: usize,
     pub latest_segment_id: usize,
+    pub ts_start: u64, // Timestamp of the first event in this segment
+    pub ts_end: u64,   // Timestamp of the last event in this segment
 }
 
 unsafe impl Zeroable for SegmentMeta {}
@@ -35,10 +35,8 @@ pub struct ReaderCursor {
 unsafe impl Zeroable for ReaderCursor {}
 unsafe impl Pod for ReaderCursor {}
 
-#[repr(C)]
-#[derive(Debug, Clone, Archive, Serialize, Deserialize, Default)]
-#[rkyv(derive(Debug))]
-#[derive(Copy)]
+#[repr(C, packed)]
+#[derive(Debug, Clone, Default, Copy)]
 pub struct MMRMeta {
     pub root_hash: [u8; 32], // Replace String with a fixed-size array to make it Copy
     pub peaks_count: usize,

src/core/meta.rs

@@ -56,6 +56,13 @@ pub fn sha_256_concat_hash(left: &[u8; 32], right: &[u8; 32]) -> [u8; 32] {
     hash.as_slice().try_into().unwrap_or_default()
 }
 
+pub fn sha_256_hash(n: Vec<u8>) -> [u8; 32] {
+    let mut sha256 = sha2::Sha256::new();
+    sha256.update(n);
+    let hash = sha256.finalize();
+    hash.as_slice().try_into().unwrap_or_default()
+}
+
 #[cfg(test)]
 mod tests {
 

src/indexing/hash.rs

@@ -112,25 +112,23 @@ mod actor_tests {
     }
 
     fn make_test_store(prefix: String) -> SegmentWriterActor {
+        let tmp = tempdir().expect("failed to create tempdir");
         // small page size so tests finish quickly, 1 page per segment to avoid rollover
         let cfg = SegmentConfig {
             page_size: 32,
             pages_per_segment: 1,
             prefix,
-            lmdb_env_path: "xaeroflux".to_string(),
+            segment_dir: tmp.path().to_string_lossy().into(),
+            lmdb_env_path: tmp.path().to_string_lossy().into(),
         };
         SegmentWriterActor::new_with_config(cfg)
     }
 
     #[test]
     fn actor_appends_to_in_memory_mmr() {
         initialize();
-        let path = "xaeroflux";
-        std::fs::create_dir_all(path).expect("couldn't create LMDB dir");
-
         // store config uses small pages; we only care about MMR here
-        let dir = tempdir().expect("failed to create tempdir");
-        let store = make_test_store(dir.path().join("mmr").display().to_string());
+        let store = make_test_store("mmr".to_string());
         let actor = MmrIndexingActor::new(Some(store), None);
 
         // Fire one event

src/indexing/storage/actors/mmr_actor.rs

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