CultNet Rust: a second type for one schema id is refused, typed

F7 (operator, 2026-09-30): "Schema aliasing sounds like a really bad idea,
I support retiring it." CultNetDocumentRegistry::register inserted into its
type and schema-id maps unconditionally, so the last binding for a schema id
won, as the TypeScript registry did before 9adea366.

- register returns Result and refuses a binding whose type already holds
  another schema id, or whose schema id another type holds, with
  cultcache_rs::SchemaConflictError: the new binding's schema id, both
  document types (a binding carries no schema name), an empty record key.
  Nothing is bound on refusal. Binding a type to the schema id it holds
  again replaces the binding.
- Call sites propagate the refusal: the cultmesh-rs registry macro, the
  interop peer example, and the tests.
- Tests renamed to what they test: raw snapshot replication hydrates the
  receiver's own type for the schema; a reactive document coalesces direct
  member writes.
- New test: both claimant shapes in both orders, the first binding intact
  afterwards, and a same-type rebind replacing.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-09-30 17:47:39 +02:00
commit fc3f3ee5ff
4 changed files with 104 additions and 29 deletions

View file

@ -62,7 +62,7 @@ macro_rules! cultmesh_documents {
$schema_version.to_string(), $schema_version.to_string(),
$schema_version.to_string(), $schema_version.to_string(),
), ),
); )?;
)* )*
Ok(()) Ok(())
} }

View file

@ -262,7 +262,7 @@ fn serve(config: PeerConfig) -> Result<()> {
cache.put(&note.document_id, &note)?; cache.put(&note.document_id, &note)?;
let mut document_registry = CultNetDocumentRegistry::new(); let mut document_registry = CultNetDocumentRegistry::new();
register_capability_bindings(&mut document_registry, &schema_registration.schema_id); register_capability_bindings(&mut document_registry, &schema_registration.schema_id)?;
let cache = Arc::new(Mutex::new(cache)); let cache = Arc::new(Mutex::new(cache));
let document_registry = Arc::new(document_registry); let document_registry = Arc::new(document_registry);
@ -384,7 +384,7 @@ fn dial(config: DialConfig) -> Result<()> {
cache.pull_all_backing_stores()?; cache.pull_all_backing_stores()?;
let mut document_registry = CultNetDocumentRegistry::new(); let mut document_registry = CultNetDocumentRegistry::new();
register_capability_bindings(&mut document_registry, &schema_registration.schema_id); register_capability_bindings(&mut document_registry, &schema_registration.schema_id)?;
let transport_name = if config.target_rudp_port.is_some() { let transport_name = if config.target_rudp_port.is_some() {
"rudp" "rudp"
@ -1227,38 +1227,39 @@ fn register_capability_entry_types(cache: &mut CultCache) -> Result<()> {
fn register_capability_bindings( fn register_capability_bindings(
document_registry: &mut CultNetDocumentRegistry, document_registry: &mut CultNetDocumentRegistry,
note_schema_id: &str, note_schema_id: &str,
) { ) -> Result<()> {
document_registry document_registry
.register(CultNetDocumentBinding::for_entry_with_schema_id::< .register(CultNetDocumentBinding::for_entry_with_schema_id::<
CultNetInteropNote, CultNetInteropNote,
>( >(
note_schema_id.to_string(), note_schema_id.to_string(),
INTEROP_SCHEMA_VERSION.to_string(), INTEROP_SCHEMA_VERSION.to_string(),
)) ))?
.register(CultNetDocumentBinding::for_entry_with_schema_id::< .register(CultNetDocumentBinding::for_entry_with_schema_id::<
CultNetInteropMutationIntent, CultNetInteropMutationIntent,
>( >(
MUTATION_INTENT_SCHEMA_ID.to_string(), MUTATION_INTENT_SCHEMA_ID.to_string(),
MUTATION_INTENT_SCHEMA_VERSION.to_string(), MUTATION_INTENT_SCHEMA_VERSION.to_string(),
)) ))?
.register(CultNetDocumentBinding::for_entry_with_schema_id::< .register(CultNetDocumentBinding::for_entry_with_schema_id::<
CultNetInteropMutationReceipt, CultNetInteropMutationReceipt,
>( >(
MUTATION_RECEIPT_SCHEMA_ID.to_string(), MUTATION_RECEIPT_SCHEMA_ID.to_string(),
MUTATION_RECEIPT_SCHEMA_VERSION.to_string(), MUTATION_RECEIPT_SCHEMA_VERSION.to_string(),
)) ))?
.register(CultNetDocumentBinding::for_entry_with_schema_id::< .register(CultNetDocumentBinding::for_entry_with_schema_id::<
CultNetInteropFireCommand, CultNetInteropFireCommand,
>( >(
FIRE_COMMAND_SCHEMA_ID.to_string(), FIRE_COMMAND_SCHEMA_ID.to_string(),
FIRE_COMMAND_SCHEMA_VERSION.to_string(), FIRE_COMMAND_SCHEMA_VERSION.to_string(),
)) ))?
.register(CultNetDocumentBinding::for_entry_with_schema_id::< .register(CultNetDocumentBinding::for_entry_with_schema_id::<
CultNetInteropFireReceipt, CultNetInteropFireReceipt,
>( >(
FIRE_RECEIPT_SCHEMA_ID.to_string(), FIRE_RECEIPT_SCHEMA_ID.to_string(),
FIRE_RECEIPT_SCHEMA_VERSION.to_string(), FIRE_RECEIPT_SCHEMA_VERSION.to_string(),
)); ))?;
Ok(())
} }
fn interaction_contracts() -> Vec<CultNetDocumentMutationContract> { fn interaction_contracts() -> Vec<CultNetDocumentMutationContract> {

View file

@ -4,6 +4,7 @@ use anyhow::anyhow;
use cultcache_rs::CultCache; use cultcache_rs::CultCache;
use cultcache_rs::CultCacheEnvelope; use cultcache_rs::CultCacheEnvelope;
use cultcache_rs::DatabaseEntry; use cultcache_rs::DatabaseEntry;
use cultcache_rs::SchemaConflictError;
use serde::Serialize; use serde::Serialize;
use serde::de::DeserializeOwned; use serde::de::DeserializeOwned;
use serde_json::Value; use serde_json::Value;
@ -129,12 +130,31 @@ impl CultNetDocumentRegistry {
Self::default() Self::default()
} }
pub fn register(&mut self, binding: CultNetDocumentBinding) -> &mut Self { /// Binds `binding`. One binding owns a document type and a schema id: a binding claiming a type or a schema id
/// another binding already holds is refused with [`SchemaConflictError`], naming the new binding's schema id and
/// both document types, and nothing is bound. Binding a type to the schema id it already holds replaces that binding.
pub fn register(&mut self, binding: CultNetDocumentBinding) -> Result<&mut Self> {
let holder = self
.bindings_by_type
.get(&binding.document_type)
.filter(|held| held.schema_id != binding.schema_id)
.or_else(|| {
self.bindings_by_schema_id
.get(&binding.schema_id)
.filter(|held| held.document_type != binding.document_type)
});
if let Some(holder) = holder {
return Err(anyhow::Error::new(SchemaConflictError {
schema_id: binding.schema_id.clone(),
schema_names: vec![holder.document_type.clone(), binding.document_type.clone()],
record_key: String::new(),
}));
}
self.bindings_by_schema_id self.bindings_by_schema_id
.insert(binding.schema_id.clone(), binding.clone()); .insert(binding.schema_id.clone(), binding.clone());
self.bindings_by_type self.bindings_by_type
.insert(binding.document_type.clone(), binding); .insert(binding.document_type.clone(), binding);
self Ok(self)
} }
pub fn binding(&self, document_type: &str) -> Option<&CultNetDocumentBinding> { pub fn binding(&self, document_type: &str) -> Option<&CultNetDocumentBinding> {

View file

@ -1930,7 +1930,7 @@ fn document_registry_advertises_mutation_contracts_with_bindings() -> Result<()>
"ghostlight.agent_state.v0".to_string(), "ghostlight.agent_state.v0".to_string(),
) )
.with_mutation_contract(contract.clone()), .with_mutation_contract(contract.clone()),
); )?;
assert_eq!(registry.mutation_contracts(), vec![contract]); assert_eq!(registry.mutation_contracts(), vec![contract]);
Ok(()) Ok(())
@ -1950,7 +1950,7 @@ fn document_registry_replicates_typed_cultcache_state() -> Result<()> {
let mut registry = CultNetDocumentRegistry::new(); let mut registry = CultNetDocumentRegistry::new();
registry.register(CultNetDocumentBinding::for_entry::< registry.register(CultNetDocumentBinding::for_entry::<
GhostlightAgentStateFixture, GhostlightAgentStateFixture,
>("ghostlight.agent_state.v0".to_string())); >("ghostlight.agent_state.v0".to_string()))?;
let mut origin = CultCache::new(); let mut origin = CultCache::new();
origin.register_entry_type::<GhostlightAgentStateFixture>()?; origin.register_entry_type::<GhostlightAgentStateFixture>()?;
@ -2007,7 +2007,7 @@ fn raw_snapshot_replication_preserves_messagepack_payload_bytes() -> Result<()>
let mut registry = CultNetDocumentRegistry::new(); let mut registry = CultNetDocumentRegistry::new();
registry.register(CultNetDocumentBinding::for_entry::< registry.register(CultNetDocumentBinding::for_entry::<
GhostlightAgentStateFixture, GhostlightAgentStateFixture,
>("ghostlight.agent_state.v0".to_string())); >("ghostlight.agent_state.v0".to_string()))?;
let mut origin = CultCache::new(); let mut origin = CultCache::new();
origin.register_entry_type::<GhostlightAgentStateFixture>()?; origin.register_entry_type::<GhostlightAgentStateFixture>()?;
@ -2053,7 +2053,7 @@ fn serve_read_only_raw_snapshot_omitted_v0_schema_list_is_no_filter_present_empt
let mut registry = CultNetDocumentRegistry::new(); let mut registry = CultNetDocumentRegistry::new();
registry.register(CultNetDocumentBinding::for_entry_with_schema_id::< registry.register(CultNetDocumentBinding::for_entry_with_schema_id::<
GhostlightAgentStateFixture, GhostlightAgentStateFixture,
>("ghostlight.agent-state".to_string(), None)); >("ghostlight.agent-state".to_string(), None))?;
let mut policy = CultNetReadOnlySnapshotPolicy::new(); let mut policy = CultNetReadOnlySnapshotPolicy::new();
policy.allow("ghostlight.agent-state", "row-1")?; policy.allow("ghostlight.agent-state", "row-1")?;
@ -2147,7 +2147,7 @@ fn serve_read_only_raw_snapshot_keeps_the_requested_record_key_order() -> Result
let mut registry = CultNetDocumentRegistry::new(); let mut registry = CultNetDocumentRegistry::new();
registry.register(CultNetDocumentBinding::for_entry_with_schema_id::< registry.register(CultNetDocumentBinding::for_entry_with_schema_id::<
GhostlightAgentStateFixture, GhostlightAgentStateFixture,
>("ghostlight.agent-state".to_string(), None)); >("ghostlight.agent-state".to_string(), None))?;
let mut policy = CultNetReadOnlySnapshotPolicy::new(); let mut policy = CultNetReadOnlySnapshotPolicy::new();
policy.allow("ghostlight.agent-state", "row-1")?; policy.allow("ghostlight.agent-state", "row-1")?;
@ -2207,10 +2207,64 @@ fn serve_read_only_raw_snapshot_keeps_the_requested_record_key_order() -> Result
} }
#[test] #[test]
fn raw_snapshot_replication_hydrates_same_schema_rust_aliases() -> Result<()> { fn document_registry_binds_one_type_per_schema_id_and_refuses_a_second_typed() -> Result<()> {
let canonical = |schema_id: &str| {
CultNetDocumentBinding::for_entry_with_schema_id::<GhostlightAgentStateFixture>(schema_id.to_string(), None)
};
let ui = |schema_id: &str| {
CultNetDocumentBinding::for_entry_with_schema_id::<GhostlightAgentStateUiFixture>(schema_id.to_string(), None)
};
let canonical_type = GhostlightAgentStateFixture::TYPE.to_string();
let ui_type = GhostlightAgentStateUiFixture::TYPE.to_string();
// Another type for a held schema id, and a held type for another schema id, in both registration orders.
let cases = [
(canonical("id.agent"), ui("id.agent"), "id.agent", [&canonical_type, &ui_type]),
(ui("id.agent"), canonical("id.agent"), "id.agent", [&ui_type, &canonical_type]),
(canonical("id.agent"), canonical("id.other"), "id.other", [&canonical_type, &canonical_type]),
(canonical("id.other"), canonical("id.agent"), "id.agent", [&canonical_type, &canonical_type]),
];
for (first, second, schema_id, names) in cases {
let mut registry = CultNetDocumentRegistry::new();
registry.register(first.clone())?;
let error = registry.register(second.clone()).err().expect("a second claimant is refused");
let conflict = error
.downcast_ref::<cultcache_rs::SchemaConflictError>()
.unwrap_or_else(|| panic!("{error:#}"));
assert_eq!(conflict.schema_id, schema_id);
assert_eq!(conflict.schema_names, names.map(|name| name.clone()).to_vec());
assert_eq!(conflict.record_key, "");
assert!(error.to_string().contains(schema_id), "{error:#}");
// The refused binding left nothing behind: the first binding still answers for its type and its schema id.
assert_eq!(registry.binding(&first.document_type).map(|held| held.schema_id.as_str()), Some(first.schema_id.as_str()));
assert_eq!(
registry.binding_by_schema_id(&first.schema_id).map(|held| held.document_type.as_str()),
Some(first.document_type.as_str())
);
if second.schema_id != first.schema_id {
assert!(registry.binding_by_schema_id(&second.schema_id).is_none());
}
}
// A type bound again to the schema id it holds replaces its binding.
let mut registry = CultNetDocumentRegistry::new();
registry.register(canonical("id.agent"))?;
registry.register(CultNetDocumentBinding::for_entry_with_schema_id::<GhostlightAgentStateFixture>(
"id.agent".to_string(),
Some("ghostlight.agent_state.v2".to_string()),
))?;
assert_eq!(
registry.binding(&canonical_type).and_then(|held| held.payload_schema_version.as_deref()),
Some("ghostlight.agent_state.v2")
);
Ok(())
}
#[test]
fn raw_snapshot_replication_hydrates_the_receivers_own_type_for_the_schema() -> Result<()> {
let temp = tempfile::tempdir()?; let temp = tempfile::tempdir()?;
let origin_store = temp.path().join("origin-alias.msgpack"); let origin_store = temp.path().join("origin.msgpack");
let target_store = temp.path().join("target-alias.msgpack"); let target_store = temp.path().join("target.msgpack");
let canonical = GhostlightAgentStateFixture { let canonical = GhostlightAgentStateFixture {
schema_version: "ghostlight.agent_state.v0".to_string(), schema_version: "ghostlight.agent_state.v0".to_string(),
agent_id: "epiphany.persona".to_string(), agent_id: "epiphany.persona".to_string(),
@ -2223,7 +2277,7 @@ fn raw_snapshot_replication_hydrates_same_schema_rust_aliases() -> Result<()> {
>( >(
"ghostlight.agent_state.v0".to_string(), "ghostlight.agent_state.v0".to_string(),
"ghostlight.agent_state.v0".to_string(), "ghostlight.agent_state.v0".to_string(),
)); ))?;
let mut origin = CultCache::new(); let mut origin = CultCache::new();
origin.register_entry_type::<GhostlightAgentStateFixture>()?; origin.register_entry_type::<GhostlightAgentStateFixture>()?;
origin.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&origin_store))?; origin.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&origin_store))?;
@ -2231,24 +2285,24 @@ fn raw_snapshot_replication_hydrates_same_schema_rust_aliases() -> Result<()> {
origin.put("epiphany.persona", &canonical)?; origin.put("epiphany.persona", &canonical)?;
let raw_snapshot = origin_registry.create_raw_snapshot_response( let raw_snapshot = origin_registry.create_raw_snapshot_response(
&origin, &origin,
"raw-snapshot-alias", "raw-snapshot-receiver-type",
Some(&["ghostlight.agent_state.v0".to_string()]), Some(&["ghostlight.agent_state.v0".to_string()]),
Some(&["epiphany.persona".to_string()]), Some(&["epiphany.persona".to_string()]),
)?; )?;
let mut alias_registry = CultNetDocumentRegistry::new(); let mut target_registry = CultNetDocumentRegistry::new();
alias_registry.register(CultNetDocumentBinding::for_entry_with_schema_id::< target_registry.register(CultNetDocumentBinding::for_entry_with_schema_id::<
GhostlightAgentStateUiFixture, GhostlightAgentStateUiFixture,
>( >(
"ghostlight.agent_state.v0".to_string(), "ghostlight.agent_state.v0".to_string(),
"ghostlight.agent_state.v0".to_string(), "ghostlight.agent_state.v0".to_string(),
)); ))?;
let mut target = CultCache::new(); let mut target = CultCache::new();
target.register_entry_type::<GhostlightAgentStateUiFixture>()?; target.register_entry_type::<GhostlightAgentStateUiFixture>()?;
target.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&target_store))?; target.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&target_store))?;
target.pull_all_backing_stores()?; target.pull_all_backing_stores()?;
let applied = alias_registry let applied = target_registry
.sync_raw_document_from_snapshot_response::<GhostlightAgentStateUiFixture>( .sync_raw_document_from_snapshot_response::<GhostlightAgentStateUiFixture>(
&mut target, &mut target,
&raw_snapshot, &raw_snapshot,
@ -2271,16 +2325,16 @@ fn raw_snapshot_replication_hydrates_same_schema_rust_aliases() -> Result<()> {
} }
#[test] #[test]
fn reactive_document_coalesces_direct_same_schema_alias_member_writes() -> Result<()> { fn reactive_document_coalesces_direct_member_writes() -> Result<()> {
let temp = tempfile::tempdir()?; let temp = tempfile::tempdir()?;
let target_store = temp.path().join("target-reactive-alias.msgpack"); let target_store = temp.path().join("target-reactive.msgpack");
let mut registry = CultNetDocumentRegistry::new(); let mut registry = CultNetDocumentRegistry::new();
registry.register(CultNetDocumentBinding::for_entry_with_schema_id::< registry.register(CultNetDocumentBinding::for_entry_with_schema_id::<
GhostlightReactiveNoteUiFixture, GhostlightReactiveNoteUiFixture,
>( >(
"ghostlight.reactive_note.v0".to_string(), "ghostlight.reactive_note.v0".to_string(),
"ghostlight.reactive_note.v0".to_string(), "ghostlight.reactive_note.v0".to_string(),
)); ))?;
let cache = Arc::new(Mutex::new(CultCache::new())); let cache = Arc::new(Mutex::new(CultCache::new()));
{ {
let mut cache = cache.lock().expect("cache mutex"); let mut cache = cache.lock().expect("cache mutex");
@ -2345,7 +2399,7 @@ fn reactive_document_tracks_canonical_reconciliation_delta() -> Result<()> {
>( >(
"ghostlight.reactive_note.v0".to_string(), "ghostlight.reactive_note.v0".to_string(),
"ghostlight.reactive_note.v0".to_string(), "ghostlight.reactive_note.v0".to_string(),
)); ))?;
let cache = Arc::new(Mutex::new(CultCache::new())); let cache = Arc::new(Mutex::new(CultCache::new()));
{ {
let mut cache = cache.lock().expect("cache mutex"); let mut cache = cache.lock().expect("cache mutex");