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

// SPDX-License-Identifier: BUSL-1.1

//! COMMIT / END / END TRANSACTION adapter.
//!
//! Thin pgwire shim over the protocol-neutral commit orchestrator in
//! `control/server/shared/session/commit.rs`: builds the pgwire Data-Plane
//! dispatch seam (keeping the materialize-freeze gate), drives `run_commit`,
//! and shapes the neutral [`CommitOutcome`] into a pgwire tag or error.

use std::future::Future;
use std::pin::Pin;

use pgwire::api::results::{Response, Tag};
use pgwire::error::{ErrorInfo, PgWireError, PgWireResult};

use crate::bridge::envelope::{ErrorCode, Response as DpResponse};
use crate::control::security::identity::AuthenticatedIdentity;
use crate::control::server::pgwire::types::error_to_sqlstate;
use crate::control::server::shared::ddl::sqlstate::error_code_to_sqlstate;
use crate::control::server::shared::session::{AbortReason, CommitOutcome, TxnDataPlane, commit};
use nodedb_physical::physical_task::PhysicalTask;

use super::super::core::NodeDbPgHandler;
use super::errors::calvin_cancelled_error;

/// pgwire Data-Plane dispatch seam for the neutral transaction orchestrator.
///
/// Wraps `dispatch_task_no_wal`, preserving its materialize-freeze gate so a
/// transaction that began before a clone freeze cannot COMMIT writes during the
/// freeze window.
pub(in crate::control::server::pgwire::handler) struct PgwireTxnDp<'a> {
    pub(in crate::control::server::pgwire::handler) handler: &'a NodeDbPgHandler,
}

impl TxnDataPlane for PgwireTxnDp<'_> {
    fn dispatch_no_wal<'a>(
        &'a self,
        task: PhysicalTask,
        wal_lsn: Option<crate::types::Lsn>,
    ) -> Pin<Box<dyn Future<Output = crate::Result<DpResponse>> + Send + 'a>> {
        Box::pin(self.handler.dispatch_task_no_wal(task, None, wal_lsn))
    }
}

impl NodeDbPgHandler {
    /// Handle COMMIT / END / END TRANSACTION.
    pub(in crate::control::server::pgwire::handler) async fn handle_commit(
        &self,
        identity: &AuthenticatedIdentity,
        addr: &std::net::SocketAddr,
    ) -> PgWireResult<Vec<Response>> {
        let dp = PgwireTxnDp { handler: self };
        match commit::run_commit(&self.sessions, addr, identity, &self.state, &dp).await {
            CommitOutcome::Committed => Ok(vec![Response::Execution(Tag::new("COMMIT"))]),
            CommitOutcome::Aborted { reason } => Err(commit_abort_to_pgerror(&reason)),
        }
    }
}

/// Map a neutral commit abort reason to the pgwire error the pre-extraction
/// COMMIT path emitted.
fn commit_abort_to_pgerror(reason: &AbortReason) -> PgWireError {
    let (severity, code, message): (&'static str, &'static str, String) = match reason {
        AbortReason::Serialization => (
            "ERROR",
            "40001",
            "could not serialize access due to concurrent update".to_owned(),
        ),
        AbortReason::NoTransaction => (
            "ERROR",
            "25000",
            "current transaction is aborted, commands ignored until end of transaction block"
                .to_owned(),
        ),
        AbortReason::BatchRejected { code } => {
            let code = code.clone().unwrap_or(ErrorCode::RejectedPrevalidation {
                reason: "transaction commit failed".to_owned(),
            });
            error_code_to_sqlstate(&code)
        }
        AbortReason::CalvinCancelled => return calvin_cancelled_error(),
        AbortReason::CalvinTimeout => (
            "ERROR",
            "57014",
            "timed out waiting for Calvin sequencer".to_owned(),
        ),
        AbortReason::Dispatch(e) | AbortReason::DdlPropose(e) => error_to_sqlstate(e),
    };
    PgWireError::UserError(Box::new(ErrorInfo::new(
        severity.to_owned(),
        code.to_owned(),
        message,
    )))
}