nodedb 0.4.0

Local-first, real-time, edge-to-cloud hybrid database for multi-modal workloads
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

// SPDX-License-Identifier: BUSL-1.1

//! Collection-schema announce helper.
//!
//! Builds the `CollectionSchema` frame that must precede the first shape
//! snapshot or delta for a collection on a sync session. The emit side of
//! the announce-precedes-data invariant: any peer that receives a delta or
//! snapshot for a collection has already received its descriptor.

use nodedb_types::sync::wire::{CollectionDescriptor, CollectionSchemaSyncMsg};

use tracing::warn;

use crate::control::state::SharedState;

use super::super::session::SyncSession;
use super::super::wire::{SyncFrame, SyncMessageType};

/// Build a `CollectionSchema` announce frame for `collection` if it has not
/// already been announced to the peer this session.
///
/// Returns `None` when the collection was already announced (idempotent) or
/// cannot be resolved in the catalog (logged; the caller then proceeds to
/// send the data frame anyway so delivery never regresses). The caller is
/// responsible for marking the collection announced only after the returned
/// frame has been sent successfully.
pub(in crate::control::server::sync) fn build_collection_schema_frame(
    shared: &SharedState,
    session: &SyncSession,
    tenant_id: u64,
    collection: &str,
) -> Option<SyncFrame> {
    if session.announced_collections.contains(collection) {
        return None;
    }

    let stored = shared
        .credentials
        .catalog()
        .get_collection(crate::types::DatabaseId::DEFAULT, tenant_id, collection)
        .ok()
        .flatten();

    let Some(stored) = stored else {
        warn!(
            session = %session.session_id,
            collection,
            tenant_id,
            "sync: collection not found in catalog; cannot announce schema before data"
        );
        return None;
    };

    let msg = CollectionSchemaSyncMsg {
        descriptor: CollectionDescriptor::from(&stored),
        // Not correctness-load-bearing: the receive side materializes the
        // collection create-only and does not consume this timestamp.
        creation_hlc: nodedb_types::hlc::Hlc::ZERO,
    };

    SyncFrame::new_msgpack(SyncMessageType::CollectionSchema, &msg)
}