cultcache-rs: expose a store's schema catalog entries (version, hash, compatible ids) and the decoder's id lookup

SingleFileMessagePackBackingStore::pull_schema_catalog reads the catalog under the shared lock pull_all takes; resolve_schema_entry resolves a record's schema id as the decoder does. Decoder, encoder and CultCacheEnvelope unchanged.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

packages/cultcache-rs/src/lib.rs

@@ -480,6 +480,39 @@ struct PersistedSchemaCatalogEntry(
     Vec<PersistedSchemaCatalogMember>,
 );
 
+/// One entry of a store file's schema catalog, read-only: the scalar fields every runtime publishes for a schema id. The members and
+/// the canonical schema JSON the file also carries are not exposed. `schema_version` is the version string the writer published
+/// (C# `CultSchemaCatalogEntry.SchemaVersion`).
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct SchemaCatalogEntry {
+    pub schema_id: String,
+    pub schema_name: String,
+    pub schema_version: String,
+    pub content_hash: String,
+    pub compatible_schema_ids: Vec<String>,
+}
+
+impl From<&PersistedSchemaCatalogEntry> for SchemaCatalogEntry {
+    fn from(entry: &PersistedSchemaCatalogEntry) -> Self {
+        Self {
+            schema_id: entry.0.clone(),
+            schema_name: entry.1.clone(),
+            schema_version: entry.2.clone(),
+            content_hash: entry.3.clone(),
+            compatible_schema_ids: entry.5.clone(),
+        }
+    }
+}
+
+/// The catalog entry a record's schema id names, as the decoder resolves it: the first entry that has the id as its own, else the
+/// first entry that lists it as compatible; `None` when the catalog publishes neither.
+pub fn resolve_schema_entry<'a>(catalog: &'a [SchemaCatalogEntry], schema_id: &str) -> Option<&'a SchemaCatalogEntry> {
+    catalog
+        .iter()
+        .find(|entry| entry.schema_id == schema_id)
+        .or_else(|| catalog.iter().find(|entry| entry.compatible_schema_ids.iter().any(|id| id == schema_id)))
+}
+
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
 struct PersistedSchemaCatalogMember(
     u32,
@@ -1697,6 +1730,16 @@ impl SingleFileMessagePackBackingStore {
         self.read_all_unlocked()
     }
 
+    /// The schema catalog the store file publishes, read under the same shared lock [`CacheBackingStore::pull_all`] takes. An absent
+    /// file and a legacy envelope array have no catalog and give an empty one. Resolve a record's `schema_id` against it with
+    /// [`resolve_schema_entry`].
+    pub fn pull_schema_catalog(&self) -> Result<Vec<SchemaCatalogEntry>> {
+        if is_absent(&self.path)? && is_absent(&self.lock_path())? {
+            return Ok(Vec::new());
+        }
+        self.with_shared_lock(|| Ok(self.read_store_unlocked()?.catalog.iter().map(SchemaCatalogEntry::from).collect()))
+    }
+
     fn write_all_unlocked(&self, entries: &[CultCacheEnvelope]) -> Result<()> {
         let failed = |kind| CultCacheStoreWriteFailed {
             path: self.path.clone(),

packages/cultcache-rs/tests/schema_catalog.rs

@@ -0,0 +1,125 @@
+//! A consumer of a single-file store reads the schema version a record was published under from the store's catalog, through the
+//! public surface only: `pull_all`, `pull_schema_catalog` and `resolve_schema_entry`
+//! (`cultlib-gaps:follow_up:rs-snapshot-schema-version`).
+
+use std::path::PathBuf;
+
+use cultcache_rs::{
+    CacheBackingStore, CultCache, DatabaseEntry, SchemaCatalogEntry, SingleFileMessagePackBackingStore, resolve_schema_entry,
+};
+
+#[derive(Clone, Debug, PartialEq, Eq, DatabaseEntry)]
+#[cultcache(type = "tests.sc_one", schema = "tests.sc_one")]
+struct One {
+    #[cultcache(key = 0)]
+    text: String,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, DatabaseEntry)]
+#[cultcache(type = "tests.sc_two", schema = "tests.sc_two")]
+struct Two {
+    #[cultcache(key = 0)]
+    text: String,
+}
+
+fn vectors(folder: &str) -> PathBuf {
+    PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../../tests/vectors").join(folder)
+}
+
+/// The version a store publishes for each record, in key order: `(key, schema name, schema version)`.
+fn versions(store: &SingleFileMessagePackBackingStore) -> Vec<(String, String, String)> {
+    let catalog = store.pull_schema_catalog().unwrap();
+    let mut rows: Vec<_> = store
+        .pull_all()
+        .unwrap()
+        .iter()
+        .map(|envelope| {
+            let id = envelope.schema_id.as_deref().unwrap();
+            let entry = resolve_schema_entry(&catalog, id).unwrap_or_else(|| panic!("no entry for {id}"));
+            (envelope.key.clone(), entry.schema_name.clone(), entry.schema_version.clone())
+        })
+        .collect();
+    rows.sort();
+    rows
+}
+
+fn copy(from: PathBuf) -> (tempfile::TempDir, SingleFileMessagePackBackingStore) {
+    let dir = tempfile::tempdir().unwrap();
+    let path = dir.path().join("store.cc");
+    std::fs::copy(from, &path).unwrap();
+    (dir, SingleFileMessagePackBackingStore::new(&path))
+}
+
+#[test]
+fn a_consumer_reads_the_version_of_each_type_a_rust_store_holds() {
+    let dir = tempfile::tempdir().unwrap();
+    let path = dir.path().join("store.cc");
+    let mut cache = CultCache::new();
+    cache.register_entry_type::<One>().unwrap();
+    cache.register_entry_type::<Two>().unwrap();
+    cache.add_backing_store(SingleFileMessagePackBackingStore::new(&path), ["tests.sc_one", "tests.sc_two"]).unwrap();
+    cache.pull_all_backing_stores().unwrap();
+    cache.put("a", &One { text: "1".into() }).unwrap();
+    cache.put("b", &Two { text: "2".into() }).unwrap();
+    drop(cache);
+
+    let consumer = SingleFileMessagePackBackingStore::new(&path);
+    assert_eq!(
+        versions(&consumer),
+        vec![
+            ("a".to_string(), "tests.sc_one".to_string(), "tests.sc_one.v1".to_string()),
+            ("b".to_string(), "tests.sc_two".to_string(), "tests.sc_two.v1".to_string()),
+        ]
+    );
+}
+
+#[test]
+fn an_id_only_an_entry_lists_as_compatible_resolves_to_that_entry() {
+    let (_dir, store) = copy(vectors("document-variants-c2a/readability/compatible-id-only-v3.bin"));
+    let catalog = store.pull_schema_catalog().unwrap();
+    assert_eq!(catalog.len(), 1);
+    assert_eq!(catalog[0].schema_id, "vectors.item.next");
+    assert_eq!(catalog[0].compatible_schema_ids, vec!["vectors.item".to_string()]);
+    let entry = resolve_schema_entry(&catalog, "vectors.item").unwrap();
+    assert_eq!(entry.schema_id, "vectors.item.next");
+    assert_eq!(entry.schema_version, "vectors.item.v1");
+    assert_eq!(resolve_schema_entry(&catalog, "vectors.nowhere"), None);
+}
+
+#[test]
+fn an_id_an_entry_owns_resolves_to_its_owner_not_to_an_entry_listing_it() {
+    for vector in ["own-id-over-compatible-v3.bin", "compatible-before-owner-v3.bin"] {
+        let (_dir, store) = copy(vectors("document-variants-c2a/readability").join(vector));
+        let catalog = store.pull_schema_catalog().unwrap();
+        assert_eq!(catalog.len(), 2, "{vector}");
+        let entry = resolve_schema_entry(&catalog, "vectors.item").unwrap();
+        assert_eq!((entry.schema_id.as_str(), entry.schema_version.as_str()), ("vectors.item", "vectors.item.v1"), "{vector}");
+    }
+}
+
+#[test]
+fn a_legacy_array_and_an_absent_file_publish_no_catalog() {
+    let (_dir, store) = copy(vectors("document-variants-c2a/readability/legacy-envelopes.bin"));
+    assert!(!store.pull_all().unwrap().is_empty());
+    assert_eq!(store.pull_schema_catalog().unwrap(), Vec::<SchemaCatalogEntry>::new());
+
+    let dir = tempfile::tempdir().unwrap();
+    let absent = SingleFileMessagePackBackingStore::new(dir.path().join("none.cc"));
+    assert_eq!(absent.pull_schema_catalog().unwrap(), Vec::<SchemaCatalogEntry>::new());
+    assert!(!dir.path().join("none.cc").exists() && !dir.path().join("none.cc.lock").exists(), "reading an absent store creates nothing");
+}
+
+// `tests/vectors/key-identity/foreign-store.cc` is written by the C# reference: two types, each published with its version.
+#[test]
+fn a_store_the_csharp_reference_wrote_gives_its_versions() {
+    let (_dir, store) = copy(vectors("key-identity/foreign-store.cc"));
+    assert_eq!(
+        versions(&store),
+        vec![
+            ("j".to_string(), "tests.ki_alpha".to_string(), "tests.ki_alpha.v1".to_string()),
+            ("k".to_string(), "tests.ki_gamma".to_string(), "tests.ki_gamma.v1".to_string()),
+        ]
+    );
+    let catalog = store.pull_schema_catalog().unwrap();
+    assert_eq!(catalog[0].content_hash, "sha256:02f0c8b7ea07c11705c1365fc29136ada08c8076209cacfb8d5526229461ce7a");
+}