nodedb 0.3.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

// SPDX-License-Identifier: BUSL-1.1

//! Dependent-read barrier for Calvin scheduler.
//!
//! When a dependent-read Calvin txn arrives at an active vshard (one with
//! write keys), the scheduler inserts a [`PendingDependentBarrier`] and waits
//! for all passive vshards to deliver their read results via the per-vshard
//! Raft log (`ReplicatedWrite::CalvinReadResult`).
//!
//! Once all passive vshards have delivered, the barrier assembles the
//! `injected_reads` map and dispatches `MetaOp::CalvinExecuteActive`.
//!
//! # Determinism
//!
//! All maps here use `BTreeMap`/`BTreeSet` — never `HashMap`/`HashSet`.
//!
//! # Timing
//!
//! `Instant::now()` is used for `timeout_at` (observability / off-WAL path
//! only; barrier timeouts are never written to the WAL).

use std::collections::{BTreeMap, BTreeSet};
use std::time::Instant;

use nodedb_cluster::calvin::types::SequencedTxn;

use crate::control::cluster::calvin::scheduler::lock_manager::LockKey;
use crate::types::TenantId;
use nodedb_physical::physical_plan::meta::PassiveReadKeyId;
use nodedb_types::Value;

// ── ReadResultEvent ────────────────────────────────────────────────────────────

/// A committed `CalvinReadResult` Raft entry delivered to the active vshard's
/// scheduler.
///
/// Produced by the apply loop when a `ReplicatedWrite::CalvinReadResult`
/// entry commits on a vshard whose scheduler has an active dependent barrier.
#[derive(Debug)]
pub struct ReadResultEvent {
    /// Sequencer epoch of the txn.
    pub epoch: u64,
    /// Position within the epoch batch.
    pub position: u32,
    /// Vshard that performed the read.
    pub passive_vshard: u32,
    /// Tenant scope.
    pub tenant_id: TenantId,
    /// Decoded read values from the passive participant.
    pub values: Vec<(PassiveReadKeyId, Value)>,
}

// ── PendingDependentBarrier ────────────────────────────────────────────────────

/// A dependent-read barrier: tracks which passive vshards have delivered
/// their read results and assembles the `injected_reads` map once all are in.
pub struct PendingDependentBarrier {
    /// Original sequenced transaction.
    pub txn: SequencedTxn,
    /// Pre-computed lock key set held by this barrier.
    pub keys: BTreeSet<LockKey>,
    /// Wall-clock time at lock acquisition (for latency metrics).
    ///
    /// `Instant::now()` used for observability only; never WAL-influencing.
    pub lock_acquired_time: Instant,
    /// Passive vshards still waiting to deliver their read results.
    /// Entries are removed as `ReadResultEvent`s arrive.
    /// `BTreeSet` for determinism.
    pub waiting_for: BTreeSet<u32>,
    /// Received read values, keyed by passive vshard.
    /// `BTreeMap` for determinism.
    pub received: BTreeMap<u32, Vec<(PassiveReadKeyId, Value)>>,
    /// Deadline for passive participant delivery.
    ///
    /// `Instant::now()` used for barrier timeout (observability / off-WAL path).
    pub timeout_at: Instant,
}

impl PendingDependentBarrier {
    /// Whether all passive participants have delivered.
    pub fn is_complete(&self) -> bool {
        self.waiting_for.is_empty()
    }

    /// Assemble the full `injected_reads` map from all received values.
    ///
    /// `BTreeMap` ensures deterministic iteration order across replicas.
    pub fn assemble_injected_reads(&self) -> BTreeMap<PassiveReadKeyId, Value> {
        let mut out = BTreeMap::new();
        for values in self.received.values() {
            for (key_id, value) in values {
                out.insert(key_id.clone(), value.clone());
            }
        }
        out
    }

    /// Whether this barrier has timed out.
    ///
    /// `Instant::now()` used for timeout check (observability; off-WAL path).
    pub fn is_timed_out(&self) -> bool {
        Instant::now() > self.timeout_at // no-determinism: scheduler barrier timeout check, not Calvin WAL data
    }
}