nodedb-lite 0.1.0

Embedded edge client for NodeDB — offline vector search, graph traversal, and CRDT sync
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80

//! Schema CRDT snapshot import handler.

use nodedb_array::sync::hlc::Hlc;
use nodedb_types::sync::wire::array::ArraySchemaSyncMsg;

use crate::error::LiteError;
use crate::storage::engine::StorageEngine;

use super::dispatcher::ArrayInbound;
use super::outcome::InboundOutcome;

impl<S: StorageEngine> ArrayInbound<S> {
    /// Import a schema CRDT snapshot from Origin.
    ///
    /// Updates the local [`crate::sync::array::SchemaRegistry`] with the Loro
    /// snapshot payload. After import, ops targeting this array that
    /// previously returned `SchemaTooNew` may succeed on retry.
    pub async fn handle_schema(
        &self,
        msg: &ArraySchemaSyncMsg,
    ) -> Result<InboundOutcome, LiteError> {
        let schema_hlc = Hlc::from_bytes(&msg.schema_hlc_bytes);
        self.schemas
            .import_snapshot(&msg.array, &msg.snapshot_payload, schema_hlc)
            .await?;
        Ok(InboundOutcome::SchemaImported)
    }
}

#[cfg(test)]
mod tests {
    use std::sync::Arc;

    use nodedb_types::sync::wire::array::ArraySchemaSyncMsg;

    use crate::storage::pagedb_storage::PagedbStorageMem;
    use crate::sync::array::replica_state::ReplicaState;
    use crate::sync::array::schema_registry::SchemaRegistry;

    use super::super::fixtures::{make_inbound, simple_schema};
    use super::super::outcome::InboundOutcome;

    #[tokio::test(flavor = "multi_thread")]
    async fn handle_schema_imports() {
        let (inbound, schemas, _pending, _storage) = make_inbound().await;

        // Create a schema on a "remote" registry.
        let remote_storage = Arc::new(PagedbStorageMem::open_in_memory().await.unwrap());
        let remote_replica = Arc::new(ReplicaState::load_or_init(&*remote_storage).await.unwrap());
        let remote_schemas =
            SchemaRegistry::new(Arc::clone(&remote_storage), Arc::clone(&remote_replica));
        remote_schemas
            .put_schema("remote_arr", &simple_schema("remote_arr"))
            .await
            .unwrap();
        let snapshot_payload = remote_schemas
            .export_snapshot("remote_arr")
            .unwrap()
            .unwrap();
        let remote_hlc = remote_schemas.schema_hlc("remote_arr").unwrap();

        assert!(
            schemas.schema_hlc("remote_arr").is_none(),
            "not yet imported"
        );

        let msg = ArraySchemaSyncMsg {
            array: "remote_arr".into(),
            replica_id: 42,
            schema_hlc_bytes: remote_hlc.to_bytes(),
            snapshot_payload,
        };
        let outcome = inbound.handle_schema(&msg).await.unwrap();
        assert_eq!(outcome, InboundOutcome::SchemaImported);
        assert!(
            schemas.schema_hlc("remote_arr").is_some(),
            "must exist after import"
        );
    }
}