feat: add UserComparator::name() for stable identity persistence

src/comparator.rs

@@ -29,8 +29,9 @@ use std::sync::Arc;
 /// # Important
 ///
 /// Once a tree is created with a comparator, it must always be opened with the
-/// same comparator. Using a different comparator on an existing tree will produce
-/// incorrect results.
+/// same comparator. The comparator's [`name`](UserComparator::name) is
+/// persisted and checked on every subsequent open — a mismatch causes the open
+/// to fail with [`Error::ComparatorMismatch`](crate::Error::ComparatorMismatch).
 ///
 /// # Examples
 ///
@@ -42,6 +43,10 @@ use std::sync::Arc;
 /// struct U64Comparator;
 ///
 /// impl UserComparator for U64Comparator {
+///     fn name(&self) -> &'static str {
+///         "u64-big-endian"
+///     }
+///
 ///     fn compare(&self, a: &[u8], b: &[u8]) -> Ordering {
 ///         if a.len() == 8 && b.len() == 8 {
 ///             // Length checked, conversion cannot fail.
@@ -57,6 +62,21 @@ use std::sync::Arc;
 /// }
 /// ```
 pub trait UserComparator: Send + Sync + std::panic::RefUnwindSafe + 'static {
+    /// Returns a stable identifier for this comparator.
+    ///
+    /// The name is persisted when a tree is first created. On subsequent
+    /// opens the stored name is compared against the caller-supplied
+    /// comparator's name — a mismatch causes the open to fail, preventing
+    /// silent data corruption from using an incompatible ordering.
+    ///
+    /// Choose a name that uniquely identifies the ordering logic and will
+    /// not change across releases (e.g. `"u64-big-endian"`, `"reverse-lexicographic"`).
+    //
+    // Intentionally required (no default impl): a shared fallback name would
+    // let two distinct comparators pass the mismatch check, silently producing
+    // corrupt reads on reopen — the exact scenario this method prevents.
+    fn name(&self) -> &'static str;
+
     /// Compares two user keys, returning their ordering.
     fn compare(&self, a: &[u8], b: &[u8]) -> std::cmp::Ordering;
 
@@ -78,6 +98,10 @@ pub trait UserComparator: Send + Sync + std::panic::RefUnwindSafe + 'static {
 pub struct DefaultUserComparator;
 
 impl UserComparator for DefaultUserComparator {
+    fn name(&self) -> &'static str {
+        "default"
+    }
+
     #[inline]
     fn compare(&self, a: &[u8], b: &[u8]) -> std::cmp::Ordering {
         a.cmp(b)
@@ -92,6 +116,11 @@ impl UserComparator for DefaultUserComparator {
 /// Shared reference to a [`UserComparator`].
 pub type SharedComparator = Arc<dyn UserComparator>;
 
+/// Maximum byte length for a comparator name.
+///
+/// Enforced on write (`persist_version`) and read (`Manifest::decode_from`).
+pub const MAX_COMPARATOR_NAME_BYTES: usize = 256;
+
 /// Returns the default comparator (lexicographic byte ordering).
 ///
 /// Uses a shared static instance to avoid repeated allocations.
@@ -102,3 +131,19 @@ pub fn default_comparator() -> SharedComparator {
         std::sync::LazyLock::new(|| Arc::new(DefaultUserComparator));
     DEFAULT.clone()
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn default_comparator_name() {
+        assert_eq!(DefaultUserComparator.name(), "default");
+        assert_eq!(default_comparator().name(), "default");
+    }
+
+    #[test]
+    fn default_comparator_is_lexicographic() {
+        assert!(DefaultUserComparator.is_lexicographic());
+    }
+}

src/config/mod.rs

@@ -605,13 +605,10 @@ impl<F: Fs> Config<F> {
     ///
     /// # Important
     ///
-    /// Once a tree is created with a custom comparator, it **must** be
-    /// re-opened with the same comparator. Using a different comparator
-    /// on an existing tree produces incorrect results.
-    ///
-    /// The comparator identity is **not** persisted to disk — the caller
-    /// is responsible for ensuring the same comparator is used across
-    /// open/close cycles (same approach as `RocksDB`).
+    /// The comparator's [`crate::UserComparator::name`] is persisted when a tree is
+    /// first created. On subsequent opens the stored name is compared against
+    /// the supplied comparator's name — a mismatch causes the open to fail
+    /// with [`Error::ComparatorMismatch`](crate::Error::ComparatorMismatch).
     #[must_use]
     pub fn comparator(mut self, comparator: SharedComparator) -> Self {
         self.comparator = comparator;

src/error.rs

@@ -76,6 +76,18 @@ pub enum Error {
     /// Decryption failed
     Decrypt(&'static str),
 
+    /// Comparator mismatch on tree reopen.
+    ///
+    /// The tree was created with a comparator whose [`crate::UserComparator::name`]
+    /// differs from the one supplied at reopen time.
+    ComparatorMismatch {
+        /// Comparator name persisted in the tree metadata.
+        stored: String,
+
+        /// Comparator name supplied by the caller.
+        supplied: &'static str,
+    },
+
     /// Range tombstone block decode failure.
     RangeTombstoneDecode {
         /// Which field or validation failed (e.g. `start_len`, `start`, `seqno`, `interval`)

src/manifest.rs

@@ -14,6 +14,7 @@ pub struct Manifest {
     )]
     pub tree_type: TreeType,
     pub level_count: u8,
+    pub comparator_name: String,
 }
 
 impl Manifest {
@@ -90,10 +91,147 @@ impl Manifest {
             );
         }
 
+        // Optional section — absent in manifests written before comparator
+        // identity persistence was added. The `UserComparator` trait was
+        // introduced in the same release cycle, so all pre-existing trees
+        // used `DefaultUserComparator` whose `name()` returns "default".
+        // Custom comparators cannot exist in old manifests.
+        let comparator_name = match toc.section(b"comparator_name") {
+            Some(section) => {
+                let limit = crate::comparator::MAX_COMPARATOR_NAME_BYTES as u64;
+
+                if section.len() > limit {
+                    return Err(crate::Error::DecompressedSizeTooLarge {
+                        declared: section.len(),
+                        limit,
+                    });
+                }
+
+                let mut bytes = Vec::new();
+                section.buf_reader(path)?.read_to_end(&mut bytes)?;
+
+                String::from_utf8(bytes).map_err(|e| crate::Error::Utf8(e.utf8_error()))?
+            }
+            None => "default".to_owned(),
+        };
+
         Ok(Self {
             version,
             tree_type,
             level_count,
+            comparator_name,
         })
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use byteorder::WriteBytesExt;
+    use std::io::Write;
+
+    /// Write the mandatory manifest sections (format_version, tree_type,
+    /// level_count, filter_hash_type) into an sfa archive at `path`.
+    /// If `comparator_name` is `Some`, also writes that section.
+    fn write_test_manifest(
+        path: &std::path::Path,
+        comparator_name: Option<&str>,
+    ) -> crate::Result<()> {
+        let file = std::fs::File::create(path)?;
+        let mut writer = sfa::Writer::from_writer(std::io::BufWriter::new(file));
+
+        writer.start("format_version")?;
+        writer.write_u8(FormatVersion::V4.into())?;
+
+        writer.start("tree_type")?;
+        writer.write_u8(TreeType::Standard.into())?;
+
+        writer.start("level_count")?;
+        writer.write_u8(7)?;
+
+        writer.start("filter_hash_type")?;
+        writer.write_u8(u8::from(ChecksumType::Xxh3))?;
+
+        if let Some(name) = comparator_name {
+            writer.start("comparator_name")?;
+            writer.write_all(name.as_bytes())?;
+        }
+
+        writer.finish()?;
+        Ok(())
+    }
+
+    #[test]
+    fn manifest_without_comparator_name_defaults_to_default() -> crate::Result<()> {
+        let dir = tempfile::tempdir()?;
+        let path = dir.path().join("manifest");
+
+        write_test_manifest(&path, None)?;
+
+        let reader = sfa::Reader::new(&path)?;
+        let manifest = Manifest::decode_from(&path, &reader)?;
+        assert_eq!(manifest.comparator_name, "default");
+        Ok(())
+    }
+
+    #[test]
+    fn manifest_with_comparator_name_round_trips() -> crate::Result<()> {
+        let dir = tempfile::tempdir()?;
+        let path = dir.path().join("manifest");
+
+        write_test_manifest(&path, Some("u64-big-endian"))?;
+
+        let reader = sfa::Reader::new(&path)?;
+        let manifest = Manifest::decode_from(&path, &reader)?;
+        assert_eq!(manifest.comparator_name, "u64-big-endian");
+        Ok(())
+    }
+
+    #[test]
+    fn manifest_rejects_oversized_comparator_name() -> crate::Result<()> {
+        let dir = tempfile::tempdir()?;
+        let path = dir.path().join("manifest");
+
+        let long_name = "x".repeat(300);
+        write_test_manifest(&path, Some(&long_name))?;
+
+        let reader = sfa::Reader::new(&path)?;
+        let result = Manifest::decode_from(&path, &reader);
+        assert!(
+            matches!(result, Err(crate::Error::DecompressedSizeTooLarge { .. })),
+            "expected DecompressedSizeTooLarge"
+        );
+        Ok(())
+    }
+
+    #[test]
+    fn manifest_rejects_invalid_utf8_comparator_name() -> crate::Result<()> {
+        let dir = tempfile::tempdir()?;
+        let path = dir.path().join("manifest");
+
+        // Write manifest with invalid UTF-8 bytes in comparator_name
+        let file = std::fs::File::create(&path)?;
+        let mut writer = sfa::Writer::from_writer(std::io::BufWriter::new(file));
+
+        writer.start("format_version")?;
+        writer.write_u8(FormatVersion::V4.into())?;
+        writer.start("tree_type")?;
+        writer.write_u8(TreeType::Standard.into())?;
+        writer.start("level_count")?;
+        writer.write_u8(7)?;
+        writer.start("filter_hash_type")?;
+        writer.write_u8(u8::from(ChecksumType::Xxh3))?;
+        writer.start("comparator_name")?;
+        writer.write_all(&[0xFF, 0xFE])?;
+
+        writer.finish()?;
+
+        let reader = sfa::Reader::new(&path)?;
+        let result = Manifest::decode_from(&path, &reader);
+        assert!(
+            matches!(result, Err(crate::Error::Utf8(_))),
+            "expected Utf8 error"
+        );
+        Ok(())
+    }
+}

src/table/util.rs

@@ -377,6 +377,10 @@ mod tests {
     /// Reverse comparator to exercise the Vec-allocation slow path.
     struct ReverseComparator;
     impl crate::comparator::UserComparator for ReverseComparator {
+        fn name(&self) -> &'static str {
+            "test-reverse"
+        }
+
         fn compare(&self, a: &[u8], b: &[u8]) -> std::cmp::Ordering {
             b.cmp(a)
         }
@@ -386,6 +390,9 @@ mod tests {
     fn test_compare_prefixed_slice_custom_comparator() {
         use std::cmp::Ordering::{Equal, Greater, Less};
 
+        use crate::comparator::UserComparator as _;
+        assert_eq!(ReverseComparator.name(), "test-reverse");
+
         // With reverse comparator, "abc" > "xyz" (reversed)
         assert_eq!(
             Greater,

src/tree/inner.rs

@@ -80,7 +80,7 @@ impl TreeInner {
                 crate::TreeType::Standard
             },
         );
-        persist_version(&config.path, &version)?;
+        persist_version(&config.path, &version, config.comparator.name())?;
 
         let comparator = config.comparator.clone();
 

src/tree/mod.rs

@@ -1330,6 +1330,40 @@ impl Tree {
 
         log::info!("Recovering LSM-tree at {}", config.path.display());
 
+        // Validate manifest metadata (format version, comparator name)
+        // BEFORE recover_levels, so a rejected open is side-effect free
+        // — recover_levels loads tables and cleans up orphans.
+        // Tree type is checked after recovery (needs the Version object).
+        // NOTE: the version file is read twice (here for metadata, then inside
+        // recover_levels for table/blob data). This is intentional — metadata
+        // validation must complete before any disk-mutating recovery work.
+        {
+            let version_id = crate::version::recovery::get_current_version(&config.path)?;
+            let manifest_path = config.path.join(format!("v{version_id}"));
+            let reader = sfa::Reader::new(&manifest_path)?;
+            let manifest = Manifest::decode_from(&manifest_path, &reader)?;
+
+            if !matches!(manifest.version, FormatVersion::V3 | FormatVersion::V4) {
+                return Err(crate::Error::InvalidVersion(manifest.version.into()));
+            }
+
+            let supplied_name = config.comparator.name();
+            if manifest.comparator_name != supplied_name {
+                log::warn!(
+                    "Comparator mismatch: tree was created with {:?} but opened with {:?}",
+                    manifest.comparator_name,
+                    supplied_name,
+                );
+                return Err(crate::Error::ComparatorMismatch {
+                    stored: manifest.comparator_name,
+                    supplied: supplied_name,
+                });
+            }
+
+            // IMPORTANT: Restore persisted config
+            config.level_count = manifest.level_count;
+        }
+
         let tree_id = get_next_tree_id();
 
         #[cfg(feature = "metrics")]
@@ -1344,14 +1378,6 @@ impl Tree {
         )?;
 
         {
-            let manifest_path = config.path.join(format!("v{}", version.id()));
-            let reader = sfa::Reader::new(&manifest_path)?;
-            let manifest = Manifest::decode_from(&manifest_path, &reader)?;
-
-            if !matches!(manifest.version, FormatVersion::V3 | FormatVersion::V4) {
-                return Err(crate::Error::InvalidVersion(manifest.version.into()));
-            }
-
             let requested_tree_type = match config.kv_separation_opts {
                 Some(_) => crate::TreeType::Blob,
                 None => crate::TreeType::Standard,
@@ -1364,9 +1390,6 @@ impl Tree {
                 );
                 return Err(crate::Error::Unrecoverable);
             }
-
-            // IMPORTANT: Restore persisted config
-            config.level_count = manifest.level_count;
         }
 
         let highest_table_id = version

src/version/mod.rs

@@ -631,6 +631,7 @@ impl Version {
     pub(crate) fn encode_into(
         &self,
         writer: &mut sfa::Writer<impl std::io::Write + std::io::Seek>,
+        comparator_name: &str,
     ) -> Result<(), crate::Error> {
         use crate::FormatVersion;
         use byteorder::{LittleEndian, WriteBytesExt};
@@ -659,6 +660,9 @@ impl Version {
         writer.start("filter_hash_type")?;
         writer.write_u8(u8::from(ChecksumType::Xxh3))?;
 
+        writer.start("comparator_name")?;
+        writer.write_all(comparator_name.as_bytes())?;
+
         //
         // Levels
         //