nodedb 0.2.1

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
117
118
119
120
121
122
123
124
125
126
127
128
129
130

// SPDX-License-Identifier: BUSL-1.1

//! Cluster readiness gate: raft election wait, catalog sanity check, peer warm-up.

use std::sync::Arc;
use std::time::Duration;

use tracing::info;

use crate::control::startup::ReadyGate;
use crate::control::state::SharedState;

/// All readiness gates passed to [`await_cluster_ready`].
pub struct ClusterReadyGates {
    pub raft_gate: ReadyGate,
    pub schema_gate: ReadyGate,
    pub sanity_gate: ReadyGate,
    pub data_groups_gate: ReadyGate,
    pub transport_gate: ReadyGate,
    pub warm_peers_gate: ReadyGate,
    pub health_loop_gate: ReadyGate,
    pub gateway_enable_gate: ReadyGate,
}

/// Wait for the metadata raft group to be ready, run catalog sanity checks,
/// warm the QUIC peer cache, and fire the remaining startup gates.
///
/// In single-node mode `raft_ready_rx` is `None` and the raft-ready wait is
/// skipped. Gate fires are always performed regardless of cluster mode.
pub async fn await_cluster_ready(
    shared: &Arc<SharedState>,
    raft_ready_rx: Option<tokio::sync::watch::Receiver<bool>>,
    gates: ClusterReadyGates,
) -> anyhow::Result<()> {
    let ClusterReadyGates {
        raft_gate,
        schema_gate,
        sanity_gate,
        data_groups_gate,
        transport_gate,
        warm_peers_gate,
        health_loop_gate,
        gateway_enable_gate,
    } = gates;
    // Boot-time readiness gate: in cluster mode, wait until the
    // metadata raft group has applied its first entry on this node
    // before opening any client-facing listener. This eliminates the
    // restart-window race where the first DDL would observe
    // `metadata propose: not leader` because election had not yet
    // completed.
    if let Some(mut ready_rx) = raft_ready_rx {
        const RAFT_READY_TIMEOUT: Duration = Duration::from_secs(30);
        match tokio::time::timeout(RAFT_READY_TIMEOUT, ready_rx.wait_for(|v| *v)).await {
            Ok(Ok(_)) => {
                info!("metadata raft group ready — opening client listeners");
            }
            Ok(Err(_)) => {
                raft_gate.fail("raft readiness watch dropped before signalling ready");
                return Err(anyhow::anyhow!(
                    "raft readiness watch dropped before signalling ready"
                ));
            }
            Err(_) => {
                raft_gate.fail(format!(
                    "raft readiness timeout after {RAFT_READY_TIMEOUT:?}"
                ));
                return Err(anyhow::anyhow!(
                    "raft readiness timeout after {RAFT_READY_TIMEOUT:?} — \
                     metadata group failed to apply first entry"
                ));
            }
        }
    }
    // Metadata raft group has applied its first entry (or we're
    // in single-node mode with no raft). The post-apply hooks
    // have rebuilt in-memory registries from redb.
    raft_gate.fire();
    schema_gate.fire();

    // Catalog sanity check: applied-index gate, redb
    // cross-table integrity, and in-memory registry ⇔ redb
    // verification. Any unrepairable divergence or any redb
    // integrity violation aborts startup.
    let verify_report = crate::control::cluster::verify_and_repair(shared).await?;
    if verify_report.is_acceptable() {
        info!(report = %verify_report, "catalog sanity check passed");
    } else {
        sanity_gate.fail(format!("catalog sanity check failed: {verify_report}"));
        return Err(anyhow::anyhow!(
            "catalog sanity check failed: {verify_report}"
        ));
    }
    sanity_gate.fire();
    data_groups_gate.fire();
    transport_gate.fire();

    // Warm the QUIC peer cache so the first replicated request
    // after boot doesn't pay a cold dial.
    if let (Some(transport), Some(topology)) = (
        shared.cluster_transport.as_ref(),
        shared.cluster_topology.as_ref(),
    ) {
        // Clone the topology snapshot so the read guard is dropped
        // before awaiting — clippy::await_holding_lock.
        let topo_snapshot = {
            let guard = topology.read().unwrap_or_else(|p| p.into_inner());
            guard.clone()
        };
        let warm_report = crate::control::cluster::warm_known_peers(
            transport,
            &topo_snapshot,
            shared.node_id,
            Duration::from_secs(2),
        )
        .await;
        if warm_report.attempted > 0 {
            info!(report = %warm_report, "peer cache warm-up complete");
            if !warm_report.is_complete() {
                for (id, err) in &warm_report.failed {
                    tracing::warn!(node_id = id, error = %err, "peer warm failed");
                }
            }
        }
    }
    warm_peers_gate.fire();
    health_loop_gate.fire();
    gateway_enable_gate.fire();

    Ok(())
}