Method load_all

nodedb/src/control/array_sync/schema_registry.rs:237–296

(
        db: &Database,
        _replica_id: ReplicaId,
        hlc_gen: &Arc<HlcGenerator>,
    )

    }

    fn load_all(
        db: &Database,
        _replica_id: ReplicaId,
        hlc_gen: &Arc<HlcGenerator>,
    ) -> crate::Result<HashMap<String, SchemaDoc>> {
        let txn = db.begin_read().map_err(|e| Error::Storage {
            engine: "array_sync".into(),
            detail: format!("schema_registry load_all begin_read: {e}"),
        })?;
        let table = txn.open_table(SCHEMA_DOCS).map_err(|e| Error::Storage {
            engine: "array_sync".into(),
            detail: format!("schema_registry load_all open_table: {e}"),
        })?;

        let mut docs = HashMap::new();
        let iter = table.iter().map_err(|e| Error::Storage {
            engine: "array_sync".into(),
            detail: format!("schema_registry load_all iter: {e}"),
        })?;

        for entry in iter {
            let (k, v) = entry.map_err(|e| Error::Storage {
                engine: "array_sync".into(),
                detail: format!("schema_registry load_all entry: {e}"),
            })?;
            let name = match std::str::from_utf8(k.value()) {
                Ok(s) => s.to_owned(),
                Err(e) => {
                    warn!(error = %e, "schema_registry: skipping non-UTF8 key");
                    continue;
                }
            };
            let persisted: PersistedSchema = match zerompk::from_msgpack(v.value()) {
                Ok(p) => p,
                Err(e) => {
                    warn!(name, error = %e, "schema_registry: skipping corrupt schema entry");
                    continue;
                }
            };
            let hlc_arr: [u8; 18] = match persisted.schema_hlc_bytes.try_into() {
                Ok(a) => a,
                Err(v) => {
                    warn!(
                        name,
                        len = v.len(),
                        "schema_registry: skipping entry with wrong hlc_bytes length"
                    );
                    continue;
                }
            };
            let schema_hlc = Hlc::from_bytes(&hlc_arr);
            let mut doc = SchemaDoc::new(ReplicaId::new(persisted.replica_id));
            if let Err(e) = doc.import_snapshot(&persisted.loro_snapshot, schema_hlc, hlc_gen) {
                warn!(name, error = %e, "schema_registry: skipping corrupt loro snapshot");
                continue;
            }
            docs.insert(name, doc);
        }