nodedb_lite/sync/array/inbound/
schema.rs

1//! Schema CRDT snapshot import handler.
2
3use nodedb_array::sync::hlc::Hlc;
4use nodedb_types::sync::wire::array::ArraySchemaSyncMsg;
5
6use crate::error::LiteError;
7use crate::storage::engine::StorageEngine;
8
9use super::dispatcher::ArrayInbound;
10use super::outcome::InboundOutcome;
11
12impl<S: StorageEngine> ArrayInbound<S> {
13    /// Import a schema CRDT snapshot from Origin.
14    ///
15    /// Updates the local [`crate::sync::array::SchemaRegistry`] with the Loro
16    /// snapshot payload. After import, ops targeting this array that
17    /// previously returned `SchemaTooNew` may succeed on retry.
18    pub async fn handle_schema(
19        &self,
20        msg: &ArraySchemaSyncMsg,
21    ) -> Result<InboundOutcome, LiteError> {
22        let schema_hlc = Hlc::from_bytes(&msg.schema_hlc_bytes);
23        self.schemas
24            .import_snapshot(&msg.array, &msg.snapshot_payload, schema_hlc)
25            .await?;
26        Ok(InboundOutcome::SchemaImported)
27    }
28}
29
30#[cfg(test)]
31mod tests {
32    use std::sync::Arc;
33
34    use nodedb_types::sync::wire::array::ArraySchemaSyncMsg;
35
36    use crate::storage::pagedb_storage::PagedbStorageMem;
37    use crate::sync::array::replica_state::ReplicaState;
38    use crate::sync::array::schema_registry::SchemaRegistry;
39
40    use super::super::fixtures::{make_inbound, simple_schema};
41    use super::super::outcome::InboundOutcome;
42
43    #[tokio::test(flavor = "multi_thread")]
44    async fn handle_schema_imports() {
45        let (inbound, schemas, _pending, _storage) = make_inbound().await;
46
47        // Create a schema on a "remote" registry.
48        let remote_storage = Arc::new(PagedbStorageMem::open_in_memory().await.unwrap());
49        let remote_replica = Arc::new(ReplicaState::load_or_init(&*remote_storage).await.unwrap());
50        let remote_schemas =
51            SchemaRegistry::new(Arc::clone(&remote_storage), Arc::clone(&remote_replica));
52        remote_schemas
53            .put_schema("remote_arr", &simple_schema("remote_arr"))
54            .await
55            .unwrap();
56        let snapshot_payload = remote_schemas
57            .export_snapshot("remote_arr")
58            .unwrap()
59            .unwrap();
60        let remote_hlc = remote_schemas.schema_hlc("remote_arr").unwrap();
61
62        assert!(
63            schemas.schema_hlc("remote_arr").is_none(),
64            "not yet imported"
65        );
66
67        let msg = ArraySchemaSyncMsg {
68            array: "remote_arr".into(),
69            replica_id: 42,
70            schema_hlc_bytes: remote_hlc.to_bytes(),
71            snapshot_payload,
72        };
73        let outcome = inbound.handle_schema(&msg).await.unwrap();
74        assert_eq!(outcome, InboundOutcome::SchemaImported);
75        assert!(
76            schemas.schema_hlc("remote_arr").is_some(),
77            "must exist after import"
78        );
79    }
80}
nodedb_lite/sync/array/inbound/schema.rs

nodedb_lite/sync/array/inbound/
schema.rs
