fix: [breaking] correct lexicographical ordering between topics for the topics index (#55)

Also: improved idx_context_key_range_end implementation plus some tests

Author: marvin-j97

src/api.rs

@@ -358,7 +358,7 @@ pub async fn serve(
     store: Store,
     engine: nu::Engine,
     expose: Option<String>,
-) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+) -> Result<(), BoxError> {
     if let Err(e) = store.append(
         Frame::builder("xs.start", store::ZERO_CONTEXT)
             .maybe_meta(expose.as_ref().map(|e| serde_json::json!({"expose": e})))
@@ -397,7 +397,7 @@ async fn listener_loop(
     mut listener: Listener,
     store: Store,
     engine: nu::Engine,
-) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+) -> Result<(), BoxError> {
     loop {
         let (stream, _) = listener.accept().await?;
         let io = TokioIo::new(stream);
@@ -506,9 +506,7 @@ async fn handle_import(store: &mut Store, body: hyper::body::Incoming) -> HTTPRe
         Err(e) => return response_400(format!("Invalid frame JSON: {}", e)),
     };
 
-    store
-        .insert_frame(&frame)
-        .map_err(|e| Box::new(e) as BoxError)?;
+    store.insert_frame(&frame).map_err(BoxError::from)?;
 
     Ok(Response::builder()
         .status(StatusCode::OK)

src/store/mod.rs

@@ -414,15 +414,18 @@ impl Store {
     }
 
     #[tracing::instrument(skip(self), fields(id = %id.to_string()))]
-    pub fn remove(&self, id: &Scru128Id) -> Result<(), fjall::Error> {
+    pub fn remove(&self, id: &Scru128Id) -> Result<(), crate::error::Error> {
         let Some(frame) = self.get(id) else {
             // Already deleted
             return Ok(());
         };
 
+        // Get the index topic key
+        let topic_key = idx_topic_key_from_frame(&frame)?;
+
         let mut batch = self.keyspace.batch();
         batch.remove(&self.frame_partition, id.as_bytes());
-        batch.remove(&self.idx_topic, idx_topic_key_from_frame(&frame));
+        batch.remove(&self.idx_topic, topic_key);
         batch.remove(&self.idx_context, idx_context_key_from_frame(&frame));
 
         // If this is a context frame, remove it from the contexts set
@@ -431,7 +434,8 @@ impl Store {
         }
 
         batch.commit()?;
-        self.keyspace.persist(fjall::PersistMode::SyncAll)
+        self.keyspace.persist(fjall::PersistMode::SyncAll)?;
+        Ok(())
     }
 
     pub async fn cas_reader(&self, hash: ssri::Integrity) -> cacache::Result<cacache::Reader> {
@@ -469,14 +473,19 @@ impl Store {
     }
 
     #[tracing::instrument(skip(self))]
-    pub fn insert_frame(&self, frame: &Frame) -> Result<(), fjall::Error> {
+    pub fn insert_frame(&self, frame: &Frame) -> Result<(), crate::error::Error> {
         let encoded: Vec<u8> = serde_json::to_vec(&frame).unwrap();
+
+        // Get the index topic key
+        let topic_key = idx_topic_key_from_frame(frame)?;
+
         let mut batch = self.keyspace.batch();
         batch.insert(&self.frame_partition, frame.id.as_bytes(), encoded);
-        batch.insert(&self.idx_topic, idx_topic_key_from_frame(frame), b"");
+        batch.insert(&self.idx_topic, topic_key, b"");
         batch.insert(&self.idx_context, idx_context_key_from_frame(frame), b"");
         batch.commit()?;
-        self.keyspace.persist(fjall::PersistMode::SyncAll)
+        self.keyspace.persist(fjall::PersistMode::SyncAll)?;
+        Ok(())
     }
 
     pub fn append(&self, mut frame: Frame) -> Result<Frame, crate::error::Error> {
@@ -497,6 +506,9 @@ impl Store {
             }
         }
 
+        // Check for null byte in topic (in case we're not storing the frame)
+        idx_topic_key_from_frame(&frame)?;
+
         // only store the frame if it's not ephemeral
         if frame.ttl != Some(TTL::Ephemeral) {
             self.insert_frame(&frame)?;
@@ -609,18 +621,28 @@ fn is_expired(id: &Scru128Id, ttl: &Duration) -> bool {
     now_ms >= expires_ms
 }
 
+const NULL_DELIMITER: u8 = 0;
+
 fn idx_topic_key_prefix(context_id: Scru128Id, topic: &str) -> Vec<u8> {
     let mut v = Vec::with_capacity(16 + topic.len() + 1); // context_id (16) + topic bytes + delimiter
     v.extend(context_id.as_bytes()); // binary context_id (16 bytes)
     v.extend(topic.as_bytes()); // topic string as UTF-8 bytes
-    v.push(0xFF); // delimiter
+    v.push(NULL_DELIMITER); // Delimiter for variable-sized keys
     v
 }
 
-fn idx_topic_key_from_frame(frame: &Frame) -> Vec<u8> {
+pub(crate) fn idx_topic_key_from_frame(frame: &Frame) -> Result<Vec<u8>, crate::error::Error> {
+    // Check if the topic contains a null byte when encoded as UTF-8
+    if frame.topic.as_bytes().contains(&NULL_DELIMITER) {
+        return Err(
+            "Topic cannot contain null byte (0x00) as it's used as a delimiter"
+                .to_string()
+                .into(),
+        );
+    }
     let mut v = idx_topic_key_prefix(frame.context_id, &frame.topic);
     v.extend(frame.id.as_bytes());
-    v
+    Ok(v)
 }
 
 fn idx_topic_frame_id_from_key(key: &[u8]) -> Scru128Id {
@@ -638,17 +660,12 @@ fn idx_context_key_from_frame(frame: &Frame) -> Vec<u8> {
 
 // Returns the key prefix for the next context after the given one
 fn idx_context_key_range_end(context_id: Scru128Id) -> Vec<u8> {
-    let mut bytes = context_id.as_bytes().to_vec();
-    for i in (0..bytes.len()).rev() {
-        if bytes[i] == 0xFF {
-            bytes[i] = 0;
-        } else {
-            bytes[i] += 1;
-            return bytes;
-        }
-    }
-    bytes.push(0);
-    bytes
+    let mut i = context_id.to_u128();
+
+    // NOTE: Reaching u128::MAX is probably not gonna happen...
+    i = i.saturating_add(1);
+
+    Scru128Id::from(i).as_bytes().to_vec()
 }
 
 fn deserialize_frame<B1: AsRef<[u8]>, B2: AsRef<[u8]>>(record: (B1, B2)) -> Frame {

src/store/tests.rs

@@ -23,6 +23,37 @@ mod tests_read_options {
         reencoded: Option<&'a str>,
     }
 
+    #[tokio::test]
+    async fn test_topic_index_order() {
+        let folder = tempfile::tempdir().unwrap();
+
+        let store = Store::new(folder.path().to_path_buf());
+
+        let frame1 = Frame {
+            id: scru128::new(),
+            topic: "ab".to_owned(),
+            ..Default::default()
+        };
+        let frame1 = store.append(frame1).unwrap();
+
+        let frame2 = Frame {
+            id: scru128::new(),
+            topic: "abc".to_owned(),
+            ..Default::default()
+        };
+        let frame2 = store.append(frame2).unwrap();
+
+        let keys = store.idx_topic.keys().flatten().collect::<Vec<_>>();
+
+        assert_eq!(
+            &[
+                fjall::Slice::from(idx_topic_key_from_frame(&frame1).unwrap()),
+                fjall::Slice::from(idx_topic_key_from_frame(&frame2).unwrap()),
+            ],
+            &*keys,
+        );
+    }
+
     #[tokio::test]
     async fn test_topic_index() {
         let folder = tempfile::tempdir().unwrap();
@@ -493,6 +524,32 @@ mod tests_context {
     use super::*;
     use tempfile::TempDir;
 
+    #[tokio::test]
+    async fn test_reject_null_byte_in_topic() {
+        let temp_dir = tempfile::tempdir().unwrap();
+        let store = Store::new(temp_dir.path().to_path_buf());
+
+        // Try to create a frame with a topic containing a null byte
+        let frame = Frame {
+            id: scru128::new(),
+            topic: "test\0topic".to_owned(),
+            context_id: ZERO_CONTEXT,
+            hash: None,
+            meta: None,
+            ttl: None,
+        };
+
+        // Creating the index key should fail
+        let result = idx_topic_key_from_frame(&frame);
+        assert!(result.is_err());
+        assert!(result.unwrap_err().to_string().contains("null byte"));
+
+        // Trying to append the frame should also fail
+        let result = store.append(frame);
+        assert!(result.is_err());
+        assert!(result.unwrap_err().to_string().contains("null byte"));
+    }
+
     #[tokio::test]
     async fn test_context_operations() {
         let temp_dir = TempDir::new().unwrap();
@@ -891,6 +948,68 @@ mod tests_context {
         let frames_ctx2: Vec<_> = store.iter_frames(Some(ctx2), None).collect();
         assert_eq!(frames_ctx2, vec![ctx2_frame1, ctx2_frame2]);
     }
+
+    #[test]
+    fn test_idx_context_key_range_end() {
+        // Test 1: Normal case - verify basic increment works
+        let context_id = Scru128Id::from_u128(100);
+        let next_id = Scru128Id::from_u128(101);
+        let result = idx_context_key_range_end(context_id);
+        assert_eq!(result, next_id.as_bytes().to_vec());
+
+        // Test 2: Test with a complex key that's not all 0xFF
+        let complex_id = Scru128Id::from_u128(0x8000_FFFF_0000_AAAA_1234_5678_9ABC_DEF0);
+        let expected_next = Scru128Id::from_u128(0x8000_FFFF_0000_AAAA_1234_5678_9ABC_DEF1);
+        assert_eq!(
+            idx_context_key_range_end(complex_id),
+            expected_next.as_bytes().to_vec()
+        );
+
+        // Test 3: Boundary case - near maximum value
+        let near_max = Scru128Id::from_u128(u128::MAX - 1);
+        let max = Scru128Id::from_u128(u128::MAX);
+        assert_eq!(idx_context_key_range_end(near_max), max.as_bytes().to_vec());
+
+        // Test 4: Boundary case - at maximum value (saturating_add should prevent overflow)
+        let at_max = Scru128Id::from_u128(u128::MAX);
+        assert_eq!(
+            idx_context_key_range_end(at_max),
+            at_max.as_bytes().to_vec(),
+            "When at u128::MAX, saturating_add should keep the same value"
+        );
+
+        // Test 5: Integration test - make sure it works in range queries
+        let temp_dir = tempfile::tempdir().unwrap();
+        let store = Store::new(temp_dir.path().to_path_buf());
+
+        // Create first context normally
+        let ctx1_frame = store
+            .append(Frame::builder("xs.context", ZERO_CONTEXT).build())
+            .unwrap();
+        let ctx1 = ctx1_frame.id;
+
+        // For ctx2, we need to manually create and register it
+        let ctx2 = Scru128Id::from_u128(ctx1.to_u128() + 1);
+        let ctx2_frame = Frame::builder("xs.context", ZERO_CONTEXT)
+            .id(ctx2)
+            .ttl(TTL::Forever)
+            .build();
+
+        // Manually insert the frame and register the context
+        store.insert_frame(&ctx2_frame).unwrap();
+        store.contexts.write().unwrap().insert(ctx2);
+
+        // Add frames to both contexts
+        let frame1 = store.append(Frame::builder("test", ctx1).build()).unwrap();
+        let frame2 = store.append(Frame::builder("test", ctx2).build()).unwrap();
+
+        // Test that range query correctly separates the contexts
+        let frames1: Vec<_> = store.read_sync(None, None, Some(ctx1)).collect();
+        assert_eq!(frames1, vec![frame1], "Should only return frames from ctx1");
+
+        let frames2: Vec<_> = store.read_sync(None, None, Some(ctx2)).collect();
+        assert_eq!(frames2, vec![frame2], "Should only return frames from ctx2");
+    }
 }
 
 mod tests_ttl_expire {