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
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116

// SPDX-License-Identifier: BUSL-1.1

//! `SharedState::open` + subsystem/cluster wiring + quota ceiling + login
//! rate-limit capacities — the boot phase that produces the final,
//! fully-wired `Arc<SharedState>`.

use std::sync::Arc;

use nodedb::ServerConfig;
use nodedb::bootstrap;
use nodedb::bridge::dispatch::Dispatcher;
use nodedb::control::cluster::ClusterHandle;
use nodedb::control::startup::StartupGate;
use nodedb::control::state::SharedState;

/// Resources this phase borrows from Data Plane bootstrap to build
/// `SharedState`, bundled so the call site doesn't juggle six separate
/// `Arc::clone` arguments.
pub(crate) struct SharedStateInputs<'a> {
    pub(crate) dispatcher: Dispatcher,
    pub(crate) wal: Arc<nodedb::wal::WalManager>,
    pub(crate) quiesce: Arc<nodedb::bridge::quiesce::CollectionQuiesce>,
    pub(crate) array_catalog: nodedb::control::array_catalog::ArrayCatalogHandle,
    pub(crate) quarantine_registry: Arc<nodedb::storage::quarantine::QuarantineRegistry>,
    pub(crate) governor: Arc<nodedb_mem::governor::MemoryGovernor>,
    pub(crate) system_metrics: Arc<nodedb::control::metrics::SystemMetrics>,
    pub(crate) maintenance_budget: Arc<nodedb::control::maintenance::MaintenanceBudgetTracker>,
    pub(crate) cluster_handle: Option<&'a ClusterHandle>,
    pub(crate) startup_gate: &'a Arc<StartupGate>,
    pub(crate) root_span: &'a tracing::Span,
}

/// Open `SharedState`, wire subsystems/cluster handles into it, then apply
/// the global quota ceiling and login rate-limit capacities. Pure
/// relocation of what used to be inline in `main()` between Data Plane
/// bootstrap and the post-open catalog steps.
pub(crate) fn open_and_wire_state(
    config: &ServerConfig,
    inputs: SharedStateInputs<'_>,
) -> anyhow::Result<Arc<SharedState>> {
    let SharedStateInputs {
        dispatcher,
        wal,
        quiesce,
        array_catalog,
        quarantine_registry,
        governor,
        system_metrics,
        maintenance_budget,
        cluster_handle,
        startup_gate,
        root_span,
    } = inputs;

    // Create shared state with persistent system catalog.
    let mut shared = SharedState::open(
        dispatcher,
        Arc::clone(&wal),
        &config.catalog_path(),
        &config.auth,
        config.tuning.clone(),
        Arc::clone(&quiesce),
        Arc::clone(&array_catalog),
    )?;

    // Install startup gate, wire subsystems and cluster handles into SharedState.
    bootstrap::state_wiring::wire_state(
        &mut shared,
        config,
        startup_gate,
        cluster_handle,
        bootstrap::state_wiring::SharedStateComponents {
            quarantine_registry: Arc::clone(&quarantine_registry),
            governor: Arc::clone(&governor),
            system_metrics: Arc::clone(&system_metrics),
            array_catalog: Arc::clone(&array_catalog),
            maintenance_budget: Arc::clone(&maintenance_budget),
        },
        root_span,
    )?;

    // Wire global quota ceiling from server config so `ALTER DATABASE SET QUOTA`
    // can validate the sum-of-database-quotas against the cluster's physical
    // resources. `memory_limit` and `max_connections` are the only dimensions
    // the server config currently constrains; storage and QPS pass through as
    // zero (= no ceiling) until [server.storage_limit] / [server.qps_limit]
    // land. The ALTER handler treats zero on any dimension as "skip that check".
    {
        use nodedb::control::security::catalog::GlobalQuotaCeiling;
        let mem_u64 = u64::try_from(config.server.memory_limit).unwrap_or(u64::MAX);
        let conn_u64 = u64::try_from(config.server.max_connections).unwrap_or(u64::MAX);
        shared.set_quota_ceiling(GlobalQuotaCeiling {
            max_memory_bytes: mem_u64,
            max_storage_bytes: 0,
            max_qps: 0,
            max_connections: conn_u64,
        });
    }

    // Apply login rate-limit capacities from cluster config (or defaults).
    {
        let (ip_cap, user_cap) = config
            .cluster
            .as_ref()
            .map(|c| {
                (
                    c.login_attempts_per_ip_per_min,
                    c.login_attempts_per_user_per_min,
                )
            })
            .unwrap_or((30, 10));
        shared.rate_limiter.set_login_capacities(ip_cap, user_cap);
    }

    Ok(shared)
}