saddle_db/
database.rs

1use std::{path::PathBuf, str::FromStr, sync::Arc, time::Duration};
2
3use futures_util::TryStreamExt;
4use saddle_core::{ComponentLifecycle, LifecycleFuture, OperationId, Result};
5use saddle_observability::{CallKind, Observer};
6use sqlx::{
7    Connection, MySqlPool,
8    mysql::{MySqlConnectOptions, MySqlConnection, MySqlPoolOptions},
9};
10
11use crate::{
12    CallContext, DbRow, MAX_RESULT_BYTES, Statement, Transaction, TransactionFuture, WriteResult,
13    cleanup::CleanupCoordinator,
14    error::{
15        invalid_config, map_operation_error, name_mapping_bypass, result_limit_exceeded,
16        transaction_begin_failed,
17    },
18    name_mapping::{
19        MappingStartupConfig, OperationSql, PhysicalOperationPlans, StaticLogicalTable, freeze,
20    },
21    row::row_payload_bytes,
22};
23
24pub const MAX_QUERY_ROWS: usize = 10_000;
25/// Maximum MySQL protocol packet accepted by the V1 deployment contract.
26///
27/// Pool startup rejects servers configured with a larger `max_allowed_packet`,
28/// and every physical pool connection repeats the check. This value remains
29/// below MySQL's `0xFF_FF_FF` continuation threshold, so sqlx receives one
30/// bounded buffer and cannot enter its two-fragment aggregate preallocation.
31pub const MAX_INBOUND_PACKET_BYTES: u64 = 8_388_608;
32
33/// Frozen deployment input for the sole process-wide database.
34///
35/// It contains the resolved secret value and mapping directory but exposes
36/// neither. Generated framework assembly consumes it exactly once to finish
37/// registering its static tables and operations before pool startup.
38pub struct DatabaseStartupInjection {
39    url: String,
40    mapping_directory: PathBuf,
41}
42
43#[derive(Clone, Copy, Debug, Eq, PartialEq)]
44pub enum DatabaseStartupInjectionError {
45    InvalidConnectionEnvironment,
46    MissingConnectionSecret,
47    InvalidConnectionSecret,
48    InvalidMappingDirectory,
49    MappingDirectoryUnavailable,
50}
51
52impl DatabaseStartupInjectionError {
53    pub const fn code(self) -> &'static str {
54        match self {
55            Self::InvalidConnectionEnvironment => "db.startup.connection_env_invalid",
56            Self::MissingConnectionSecret => "db.startup.connection_secret_missing",
57            Self::InvalidConnectionSecret => "db.startup.connection_secret_invalid",
58            Self::InvalidMappingDirectory => "db.startup.mapping_dir_invalid",
59            Self::MappingDirectoryUnavailable => "db.startup.mapping_dir_unavailable",
60        }
61    }
62}
63
64impl DatabaseStartupInjection {
65    /// Resolves the database portion of the single `saddle.toml` deployment
66    /// input. No database connection or metadata query is performed here.
67    #[doc(hidden)]
68    pub fn load(
69        config_directory: &std::path::Path,
70        connection_environment: &str,
71        mapping_directory: &std::path::Path,
72    ) -> std::result::Result<Self, DatabaseStartupInjectionError> {
73        if !valid_environment_name(connection_environment) {
74            return Err(DatabaseStartupInjectionError::InvalidConnectionEnvironment);
75        }
76        let url = std::env::var(connection_environment).map_err(|error| match error {
77            std::env::VarError::NotPresent => {
78                DatabaseStartupInjectionError::MissingConnectionSecret
79            }
80            std::env::VarError::NotUnicode(_) => {
81                DatabaseStartupInjectionError::InvalidConnectionSecret
82            }
83        })?;
84        if url.trim().is_empty() {
85            return Err(DatabaseStartupInjectionError::InvalidConnectionSecret);
86        }
87        if mapping_directory.as_os_str().is_empty()
88            || mapping_directory.is_absolute()
89            || mapping_directory
90                .components()
91                .any(|component| !matches!(component, std::path::Component::Normal(_)))
92        {
93            return Err(DatabaseStartupInjectionError::InvalidMappingDirectory);
94        }
95        let mapping_directory = config_directory.join(mapping_directory);
96        if !mapping_directory.is_dir() {
97            return Err(DatabaseStartupInjectionError::MappingDirectoryUnavailable);
98        }
99        Ok(Self {
100            url,
101            mapping_directory,
102        })
103    }
104
105    #[doc(hidden)]
106    pub fn into_database_config(self) -> DatabaseConfig {
107        DatabaseConfig::new(self.url).name_mapping_directory(self.mapping_directory)
108    }
109}
110
111fn valid_environment_name(value: &str) -> bool {
112    let mut bytes = value.bytes();
113    let Some(first) = bytes.next() else {
114        return false;
115    };
116    (first.is_ascii_alphabetic() || first == b'_')
117        && bytes.all(|byte| byte.is_ascii_alphanumeric() || byte == b'_')
118}
119
120/// Configuration for the single V1 MySQL/MariaDB data source.
121#[derive(Clone)]
122pub struct DatabaseConfig {
123    url: String,
124    max_connections: u32,
125    acquire_timeout: Duration,
126    name_mappings: Option<MappingStartupConfig>,
127}
128
129impl DatabaseConfig {
130    pub fn new(url: impl Into<String>) -> Self {
131        Self {
132            url: url.into(),
133            max_connections: 16,
134            acquire_timeout: Duration::from_secs(5),
135            name_mappings: None,
136        }
137    }
138    pub fn max_connections(mut self, value: u32) -> Self {
139        self.max_connections = value;
140        self
141    }
142    pub fn acquire_timeout(mut self, value: Duration) -> Self {
143        self.acquire_timeout = value;
144        self
145    }
146
147    /// Enables deployment-provided logical-to-physical name mappings.
148    ///
149    /// The directory must contain exactly one JSON file per registered table.
150    pub fn name_mapping_directory(mut self, directory: impl Into<PathBuf>) -> Self {
151        self.name_mappings
152            .get_or_insert_with(|| MappingStartupConfig::new(PathBuf::new()))
153            .set_directory(directory.into());
154        self
155    }
156
157    /// Registers one generated logical table whose complete mapping is required.
158    pub fn register_logical_table<T: StaticLogicalTable>(mut self) -> Self {
159        self.name_mappings
160            .get_or_insert_with(|| MappingStartupConfig::new(PathBuf::new()))
161            .register_table::<T>();
162        self
163    }
164
165    #[doc(hidden)]
166    pub fn register_query_operation<O: crate::internal::StaticQueryOptionalOperation>(
167        mut self,
168    ) -> Self {
169        self.name_mappings
170            .get_or_insert_with(|| MappingStartupConfig::new(PathBuf::new()))
171            .register_query::<O>(O::OPERATION, O::LOGICAL_TABLE, O::LOGICAL_COLUMNS, O::SQL);
172        self
173    }
174
175    #[doc(hidden)]
176    pub fn register_write_operation<O: crate::internal::StaticWriteOperation>(mut self) -> Self {
177        self.name_mappings
178            .get_or_insert_with(|| MappingStartupConfig::new(PathBuf::new()))
179            .register_write::<O>(O::OPERATION, O::LOGICAL_TABLE, O::LOGICAL_COLUMNS, O::SQL);
180        self
181    }
182
183    /// Performs the same local mapping validation used by startup, without
184    /// connecting to or inspecting the database server.
185    pub fn validate_name_mappings(
186        &self,
187    ) -> std::result::Result<(), crate::DatabaseNameMappingError> {
188        freeze(self.name_mappings.clone()).map(|_| ())
189    }
190
191    pub(crate) fn verified_connections(mut self, value: u32) -> Self {
192        self.max_connections = value;
193        self
194    }
195
196    pub(crate) fn deployment_url(&self) -> &str {
197        &self.url
198    }
199
200    pub(crate) fn options(&self) -> Result<MySqlConnectOptions> {
201        if self.max_connections == 0 {
202            return Err(invalid_config("max connections must be greater than zero"));
203        }
204        if self.acquire_timeout.is_zero() {
205            return Err(invalid_config("acquire timeout must be greater than zero"));
206        }
207        MySqlConnectOptions::from_str(&self.url)
208            .map_err(|_| invalid_config("database URL is not a valid MySQL/MariaDB URL"))
209    }
210}
211
212/// The process-wide managed database capability.
213#[derive(Clone)]
214pub struct Database {
215    pub(crate) pool: MySqlPool,
216    observer: Observer,
217    cleanup: Arc<CleanupCoordinator>,
218    physical_plans: Arc<PhysicalOperationPlans>,
219}
220
221impl Database {
222    /// Creates the single managed pool and verifies that it can connect.
223    pub async fn connect(config: DatabaseConfig, observer: Observer) -> Result<Self> {
224        let physical_plans = freeze(config.name_mappings.clone())
225            .map_err(|_| invalid_config("database name mapping is invalid"))?;
226        let options = config.options()?;
227        let mut preflight = MySqlConnection::connect_with(&options)
228            .await
229            .map_err(map_operation_error)?;
230        let server_packet_limit = sqlx::query_scalar::<_, u64>("SELECT @@max_allowed_packet")
231            .fetch_one(&mut preflight)
232            .await
233            .map_err(map_operation_error)?;
234        preflight.close().await.map_err(map_operation_error)?;
235        if server_packet_limit > MAX_INBOUND_PACKET_BYTES {
236            return Err(invalid_config(
237                "server max_allowed_packet exceeds the V1 inbound allocation limit",
238            ));
239        }
240        let pool = MySqlPoolOptions::new()
241            .max_connections(config.max_connections)
242            .acquire_timeout(config.acquire_timeout)
243            .idle_timeout(None)
244            .max_lifetime(None)
245            .after_connect(|connection, _metadata| {
246                Box::pin(async move {
247                    let packet_limit = sqlx::query_scalar::<_, u64>("SELECT @@max_allowed_packet")
248                        .fetch_one(connection)
249                        .await?;
250                    if packet_limit > MAX_INBOUND_PACKET_BYTES {
251                        return Err(sqlx::Error::Protocol(
252                            "server packet limit exceeds Saddle V1 allocation boundary".to_owned(),
253                        ));
254                    }
255                    Ok(())
256                })
257            })
258            .connect_with(options)
259            .await
260            .map_err(map_operation_error)?;
261        let mut preopened = Vec::new();
262        preopened
263            .try_reserve_exact(config.max_connections as usize)
264            .map_err(|_| invalid_config("database connection profile is too large"))?;
265        while preopened.len() < config.max_connections as usize {
266            preopened.push(pool.acquire().await.map_err(map_operation_error)?);
267        }
268        for connection in &mut preopened {
269            connection.return_to_pool().await;
270        }
271        Ok(Self {
272            pool,
273            observer,
274            cleanup: CleanupCoordinator::start(),
275            physical_plans: Arc::new(physical_plans),
276        })
277    }
278
279    pub(crate) fn query_sql<O: crate::internal::StaticQueryOptionalOperation>(
280        &self,
281    ) -> OperationSql {
282        self.physical_plans.query::<O>()
283    }
284
285    pub(crate) fn write_sql<O: crate::internal::StaticWriteOperation>(&self) -> OperationSql {
286        self.physical_plans.write::<O>()
287    }
288
289    pub async fn query_all(
290        &self,
291        parent: &CallContext,
292        statement: Statement,
293    ) -> Result<Vec<DbRow>> {
294        if self.physical_plans.enabled() {
295            return Err(name_mapping_bypass());
296        }
297        statement.validate()?;
298        let operation = statement.operation().to_owned();
299        let call = self.observer.start_child_call(
300            parent,
301            CallKind::Database,
302            "database",
303            "database",
304            OperationId::from(operation),
305        );
306        let result = async {
307            let mut stream = statement.query().fetch(&self.pool);
308            let mut rows = Vec::new();
309            let mut result_bytes = 0_usize;
310            while let Some(row) = stream.try_next().await.map_err(map_operation_error)? {
311                if rows.len() == MAX_QUERY_ROWS {
312                    return Err(result_limit_exceeded());
313                }
314                result_bytes = result_bytes.saturating_add(row_payload_bytes(&row)?);
315                if result_bytes > MAX_RESULT_BYTES {
316                    return Err(result_limit_exceeded());
317                }
318                rows.push(DbRow(row));
319            }
320            Ok(rows)
321        }
322        .await;
323        finish_call(call, &result);
324        result
325    }
326
327    pub async fn query_optional(
328        &self,
329        parent: &CallContext,
330        statement: Statement,
331    ) -> Result<Option<DbRow>> {
332        if self.physical_plans.enabled() {
333            return Err(name_mapping_bypass());
334        }
335        statement.validate()?;
336        let operation = statement.operation().to_owned();
337        let call = self.observer.start_child_call(
338            parent,
339            CallKind::Database,
340            "database",
341            "database",
342            OperationId::from(operation),
343        );
344        let result = statement
345            .query()
346            .fetch_optional(&self.pool)
347            .await
348            .map_err(map_operation_error)
349            .and_then(|row| {
350                row.map(|row| {
351                    row_payload_bytes(&row)?;
352                    Ok(DbRow(row))
353                })
354                .transpose()
355            });
356        finish_call(call, &result);
357        result
358    }
359
360    pub async fn write(&self, parent: &CallContext, statement: Statement) -> Result<WriteResult> {
361        if self.physical_plans.enabled() {
362            return Err(name_mapping_bypass());
363        }
364        statement.validate()?;
365        let operation = statement.operation().to_owned();
366        let call = self.observer.start_child_call(
367            parent,
368            CallKind::Database,
369            "database",
370            "database",
371            OperationId::from(operation),
372        );
373        let result = statement
374            .query()
375            .execute(&self.pool)
376            .await
377            .map(|result| WriteResult::new(result.rows_affected(), result.last_insert_id()))
378            .map_err(map_operation_error);
379        finish_call(call, &result);
380        result
381    }
382
383    /// Runs one explicit, single-level transaction.
384    ///
385    /// The callback returns a boxed async block because that is the minimal
386    /// Rust API that safely ties all operations to the borrowed transaction.
387    pub async fn transaction<T, F>(
388        &self,
389        parent: &CallContext,
390        operation: impl Into<OperationId>,
391        work: F,
392    ) -> Result<T>
393    where
394        T: Send,
395        F: for<'a> FnOnce(&'a mut Transaction) -> TransactionFuture<'a, T> + Send,
396    {
397        if self.physical_plans.enabled() {
398            return Err(name_mapping_bypass());
399        }
400        let operation = operation.into();
401        crate::statement::validate_operation(operation.as_str())?;
402        let call = self.observer.start_child_call(
403            parent,
404            CallKind::Transaction,
405            "database",
406            "database",
407            operation,
408        );
409        let cleanup = match self.cleanup.transaction_sender() {
410            Some(cleanup) => cleanup,
411            None => {
412                let error = transaction_begin_failed();
413                call.fail(&error);
414                return Err(error);
415            }
416        };
417        let begin = self.observer.start_child_call(
418            call.context(),
419            CallKind::Transaction,
420            "database",
421            "database",
422            "begin",
423        );
424        let raw = match self.pool.begin().await {
425            Ok(transaction) => {
426                begin.succeed();
427                transaction
428            }
429            Err(_) => {
430                let error = transaction_begin_failed();
431                begin.fail(&error);
432                call.fail(&error);
433                return Err(error);
434            }
435        };
436        let mut transaction = Transaction::new(raw, self.observer.clone(), call, cleanup);
437        match work(&mut transaction).await {
438            Ok(value) => transaction.commit().await.map(|()| value),
439            Err(work_error) => Err(transaction.rollback(work_error).await),
440        }
441    }
442
443    pub(crate) async fn close(&self) -> Result<()> {
444        let cleanup = self.cleanup.shutdown().await;
445        self.pool.close().await;
446        cleanup
447    }
448
449    #[cfg(test)]
450    pub(crate) fn connect_lazy(config: DatabaseConfig, observer: Observer) -> Result<Self> {
451        let physical_plans = freeze(config.name_mappings.clone())
452            .map_err(|_| invalid_config("database name mapping is invalid"))?;
453        let options = config.options()?;
454        let pool = MySqlPoolOptions::new()
455            .max_connections(config.max_connections)
456            .acquire_timeout(config.acquire_timeout)
457            .connect_lazy_with(options);
458        Ok(Self {
459            pool,
460            observer,
461            cleanup: CleanupCoordinator::start(),
462            physical_plans: Arc::new(physical_plans),
463        })
464    }
465}
466
467impl ComponentLifecycle for Database {
468    fn name(&self) -> &'static str {
469        "database"
470    }
471    fn start(&self) -> LifecycleFuture<'_> {
472        Box::pin(async { Ok(()) })
473    }
474    fn shutdown(&self) -> LifecycleFuture<'_> {
475        Box::pin(async move { self.close().await })
476    }
477}
478
479fn finish_call<T>(call: saddle_observability::ActiveCall, result: &Result<T>) {
480    match result {
481        Ok(_) => call.succeed(),
482        Err(error) => call.fail(error),
483    }
484}
485
486#[cfg(test)]
487mod tests {
488    use saddle_core::{ApplicationId, ErrorKind, ModuleId, ServiceId, SpanId, TraceId};
489    use saddle_observability::ObserverConfig;
490    use serde_json::Value;
491    use std::{
492        io,
493        sync::{Arc, Mutex},
494    };
495
496    use super::*;
497    use crate::SaddleError;
498
499    struct MappedTable;
500
501    impl crate::StaticLogicalTable for MappedTable {
502        const TABLE: &'static str = "逻辑表";
503        const COLUMNS: &'static [&'static str] = &["逻辑列"];
504    }
505
506    #[derive(Clone, Default)]
507    struct Capture(Arc<Mutex<Vec<u8>>>);
508    impl io::Write for Capture {
509        fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
510            self.0.lock().unwrap().extend_from_slice(bytes);
511            Ok(bytes.len())
512        }
513        fn flush(&mut self) -> io::Result<()> {
514            Ok(())
515        }
516    }
517    fn context() -> CallContext {
518        CallContext::new(
519            ApplicationId::from("shop"),
520            ModuleId::from("orders"),
521            ServiceId::from("orders"),
522            OperationId::from("create"),
523            TraceId::from_u128(1),
524            SpanId::from_u64(2),
525        )
526    }
527
528    #[test]
529    fn configuration_rejects_invalid_bounds_without_exposing_url() {
530        let observer = Observer::with_writer(ObserverConfig::default(), io::sink()).unwrap();
531        let error = Database::connect_lazy(
532            DatabaseConfig::new("mysql://secret@localhost/db").max_connections(0),
533            observer,
534        )
535        .err()
536        .unwrap();
537        assert_eq!(error.code(), "db.invalid_config");
538        assert!(!error.to_string().contains("secret"));
539    }
540
541    #[test]
542    fn unified_startup_injection_resolves_secret_and_relative_mapping_directory() {
543        let directory =
544            std::env::temp_dir().join(format!("saddle-db-alpha10-config-{}", std::process::id()));
545        let mappings = directory.join("mappings");
546        let _ = std::fs::remove_dir_all(&directory);
547        std::fs::create_dir_all(&mappings).unwrap();
548        let executable = std::env::current_exe().unwrap();
549        for mode in ["happy", "missing", "bad-path"] {
550            let mut child = std::process::Command::new(&executable);
551            child
552                .args([
553                    "--ignored",
554                    "--exact",
555                    "database::tests::unified_startup_injection_child",
556                ])
557                .env("SADDLE_ALPHA10_INJECTION_MODE", mode)
558                .env("SADDLE_ALPHA10_CONFIG_ROOT", &directory)
559                .env_remove("SADDLE_ALPHA10_DATABASE_TEST");
560            if mode != "missing" {
561                child.env(
562                    "SADDLE_ALPHA10_DATABASE_TEST",
563                    "mysql://deployment-secret@localhost/database",
564                );
565            }
566            assert!(
567                child.status().unwrap().success(),
568                "child mode {mode} failed"
569            );
570        }
571        std::fs::remove_dir_all(directory).unwrap();
572    }
573
574    #[test]
575    #[ignore = "executed in isolated child processes by the parent test"]
576    fn unified_startup_injection_child() {
577        let root = PathBuf::from(std::env::var_os("SADDLE_ALPHA10_CONFIG_ROOT").unwrap());
578        let mode = std::env::var("SADDLE_ALPHA10_INJECTION_MODE").unwrap();
579        let mapping = if mode == "bad-path" {
580            std::path::Path::new("../mappings")
581        } else {
582            std::path::Path::new("mappings")
583        };
584        let result = DatabaseStartupInjection::load(&root, "SADDLE_ALPHA10_DATABASE_TEST", mapping);
585        match mode.as_str() {
586            "happy" => {
587                let injection = result.unwrap();
588                assert_eq!(injection.mapping_directory, root.join("mappings"));
589                assert!(injection.url.contains("deployment-secret"));
590            }
591            "missing" => assert_eq!(
592                result.err(),
593                Some(DatabaseStartupInjectionError::MissingConnectionSecret)
594            ),
595            "bad-path" => assert_eq!(
596                result.err(),
597                Some(DatabaseStartupInjectionError::InvalidMappingDirectory)
598            ),
599            _ => panic!("unknown child mode"),
600        }
601    }
602
603    #[tokio::test]
604    async fn mapping_enabled_rejects_legacy_raw_statement_before_database_io() {
605        let directory =
606            std::env::temp_dir().join(format!("saddle-db-alpha7-bypass-{}", std::process::id()));
607        let _ = std::fs::remove_dir_all(&directory);
608        std::fs::create_dir(&directory).unwrap();
609        std::fs::write(
610            directory.join("table.json"),
611            r#"{"table":{"from":"逻辑表","to":"physical_table"},"columns":[{"from":"逻辑列","to":"physical_column"}]}"#,
612        )
613        .unwrap();
614        let observer = Observer::with_writer(ObserverConfig::default(), io::sink()).unwrap();
615        let database = Database::connect_lazy(
616            DatabaseConfig::new("mysql://localhost/unused")
617                .name_mapping_directory(&directory)
618                .register_logical_table::<MappedTable>(),
619            observer,
620        )
621        .unwrap();
622        let error = database
623            .write(
624                &context(),
625                Statement::new("legacy.write", "SELECT 1").unwrap(),
626            )
627            .await
628            .unwrap_err();
629        assert_eq!(error.code(), "db.name_mapping_required");
630        database.shutdown().await.unwrap();
631        std::fs::remove_dir_all(directory).unwrap();
632    }
633
634    #[tokio::test]
635    async fn closed_pool_query_has_stable_error_and_trace_record() {
636        let capture = Capture::default();
637        let observer = Observer::with_writer(ObserverConfig::default(), capture.clone()).unwrap();
638        let database = Database::connect_lazy(
639            DatabaseConfig::new("mysql://localhost/db"),
640            observer.clone(),
641        )
642        .unwrap();
643        database.close().await.unwrap();
644        let error = match database
645            .query_all(
646                &context(),
647                Statement::new("orders.list", "SELECT 1").unwrap(),
648            )
649            .await
650        {
651            Ok(_) => panic!("closed pool query unexpectedly succeeded"),
652            Err(error) => error,
653        };
654        assert_eq!(error.kind(), ErrorKind::Unavailable);
655        assert_eq!(error.code(), "db.connection_unavailable");
656        observer.flush().await.unwrap();
657        let output = String::from_utf8(capture.0.lock().unwrap().clone()).unwrap();
658        let records: Vec<Value> = output
659            .lines()
660            .map(|line| serde_json::from_str(line).unwrap())
661            .collect();
662        assert_eq!(records[0]["call_kind"], "database");
663        assert_eq!(records[1]["error_code"], "db.connection_unavailable");
664        assert_eq!(records[0]["trace_id"], context().trace_id().to_string());
665        assert!(!output.contains("SELECT 1"));
666    }
667
668    #[tokio::test]
669    async fn transaction_begin_failure_records_phase_and_stable_error() {
670        let capture = Capture::default();
671        let observer = Observer::with_writer(ObserverConfig::default(), capture.clone()).unwrap();
672        let database = Database::connect_lazy(
673            DatabaseConfig::new("mysql://localhost/db"),
674            observer.clone(),
675        )
676        .unwrap();
677        database.pool.close().await;
678        let error = database
679            .transaction(&context(), "orders.create", |_transaction| {
680                Box::pin(async { Ok::<_, SaddleError>(()) })
681            })
682            .await
683            .unwrap_err();
684        assert_eq!(error.code(), "db.transaction_begin_failed");
685        observer.flush().await.unwrap();
686        let output = String::from_utf8(capture.0.lock().unwrap().clone()).unwrap();
687        let records: Vec<Value> = output
688            .lines()
689            .map(|line| serde_json::from_str(line).unwrap())
690            .collect();
691        assert_eq!(records.len(), 4);
692        assert_eq!(records[0]["operation"], "orders.create");
693        assert_eq!(records[1]["operation"], "begin");
694        assert_eq!(records[2]["error_code"], "db.transaction_begin_failed");
695        assert_eq!(records[3]["error_code"], "db.transaction_begin_failed");
696        assert!(
697            records
698                .iter()
699                .all(|record| record["trace_id"] == context().trace_id().to_string())
700        );
701    }
702
703    #[allow(dead_code)]
704    async fn transaction_usage_compiles(database: &Database, context: &CallContext) -> Result<()> {
705        database
706            .transaction(context, "orders.create", |transaction| {
707                Box::pin(async move {
708                    transaction
709                        .write(
710                            Statement::new("orders.insert", "INSERT INTO orders(id) VALUES (?)")?
711                                .bind(1_u64)?,
712                        )
713                        .await?;
714                    transaction
715                        .query_optional(
716                            Statement::new("orders.find", "SELECT id FROM orders WHERE id = ?")?
717                                .bind(1_u64)?,
718                        )
719                        .await?;
720                    Ok(())
721                })
722            })
723            .await
724    }
725}
saddle_db/database.rs

saddle_db/
database.rs
