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
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88

// SPDX-License-Identifier: BUSL-1.1

//! Apply tenant catalog entries to `SystemCatalog` redb.

use tracing::warn;

use crate::control::security::catalog::{
    StoredCollection, StoredTenant, StoredUser, SystemCatalog,
};
use crate::types::DatabaseId;

pub fn put(stored: &StoredTenant, catalog: &SystemCatalog) {
    if let Err(e) = catalog.put_tenant(stored) {
        warn!(
            tenant = stored.tenant_id,
            name = %stored.name,
            error = %e,
            "catalog_entry: put_tenant failed"
        );
    }
}

pub fn put_with_admin(tenant: &StoredTenant, admin: &StoredUser, catalog: &SystemCatalog) -> bool {
    match catalog.put_tenant_with_admin(tenant, admin) {
        Ok(()) => true,
        Err(error) => {
            warn!(
                tenant = tenant.tenant_id,
                name = %tenant.name,
                admin = %admin.username,
                %error,
                "catalog_entry: put_tenant_with_admin failed"
            );
            false
        }
    }
}

pub fn delete(tenant_id: u64, catalog: &SystemCatalog) {
    if let Err(e) = catalog.delete_tenant(tenant_id) {
        warn!(
            tenant = tenant_id,
            error = %e,
            "catalog_entry: delete_tenant failed"
        );
    }
}

/// Apply `MoveTenantCutover`: atomically re-key all `collections` from
/// `source_db_id` to `target_db_id`, then delete each one from the source.
///
/// This is the single Raft proposal that makes the cutover phase of
/// `MOVE TENANT` atomic on every node.
pub fn move_cutover(
    tenant_id: u64,
    source_db_id: u64,
    target_db_id: u64,
    collections: &[StoredCollection],
    catalog: &SystemCatalog,
) {
    let src = DatabaseId::new(source_db_id);
    let tgt = DatabaseId::new(target_db_id);

    for coll in collections {
        // Write to target database.
        let mut target_coll = coll.clone();
        target_coll.database_id = tgt;
        if let Err(e) = catalog.put_collection(tgt, &target_coll) {
            warn!(
                tenant = tenant_id,
                collection = %coll.name,
                error = %e,
                "move_cutover: put_collection to target failed"
            );
            continue;
        }
        // Delete from source database using the collection's own tenant_id,
        // which is the actual storage key component.
        if let Err(e) = catalog.delete_collection(src, coll.tenant_id, &coll.name) {
            warn!(
                tenant = coll.tenant_id,
                collection = %coll.name,
                error = %e,
                "move_cutover: delete_collection from source failed"
            );
        }
    }
}