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

// SPDX-License-Identifier: BUSL-1.1

//! Read-reservation release and abort teardown for graceful transaction exits.

use std::net::SocketAddr;

use nodedb_cluster::calvin::types::ReleaseReason;

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

use super::store::SessionStore;

/// Drain and release every read reservation this session holds, routing one
/// sequenced `ReleaseReservation` per distinct vShard. Best-effort: a failed
/// release is swallowed by `release_reservation` (lease GC backstops). Call on
/// every graceful txn exit BEFORE the session state is cleared.
pub(super) async fn release_session_reservations(
    state: &SharedState,
    sessions: &SessionStore,
    addr: &SocketAddr,
    reason: ReleaseReason,
) {
    let (owner, vshards) = sessions.take_reservations(addr);
    let Some(owner) = owner else { return };
    for vshard in vshards {
        let _ = crate::control::planner::calvin::reservation::release_reservation(
            state, owner, vshard, reason,
        )
        .await;
    }
}

/// Standard COMMIT/abort teardown: release the transaction's read reservations
/// (with `ReleaseReason::Abort`) while the reservation owner is still set, THEN
/// roll the session back to `Idle` and free any pending GAP_FREE sequence
/// reservations. Ordering matters — `rollback_with_gap_free` clears the owner,
/// so the release must run first. Used by every COMMIT abort branch that must
/// leave the session idle without persisting; the transport adapters map
/// `Aborted` to a wire error and never roll back afterward, so each abort branch
/// owns its rollback.
pub(super) async fn release_and_rollback(
    state: &SharedState,
    sessions: &SessionStore,
    addr: &SocketAddr,
) {
    release_session_reservations(state, sessions, addr, ReleaseReason::Abort).await;
    rollback_with_gap_free(sessions, addr, state);
}

/// Roll the session back to `Idle` and release any pending GAP_FREE sequence
/// reservations.
fn rollback_with_gap_free(sessions: &SessionStore, addr: &SocketAddr, state: &SharedState) {
    if let Ok(reservations) = sessions.rollback(addr) {
        for handle in &reservations {
            let key = handle.sequence_key.clone();
            let registry = &state.sequence_registry;
            registry.gap_free_manager().rollback(handle, || {
                let map = registry.sequences_read();
                if let Some(h) = map.get(&key) {
                    h.rollback_one();
                }
            });
        }
    }
}