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(crate) fn name_mappings_frozen(&self) -> bool {
290        self.physical_plans.enabled()
291    }
292
293    pub(crate) fn observer(&self) -> &Observer {
294        &self.observer
295    }
296
297    pub async fn query_all(
298        &self,
299        parent: &CallContext,
300        statement: Statement,
301    ) -> Result<Vec<DbRow>> {
302        if self.physical_plans.enabled() {
303            return Err(name_mapping_bypass());
304        }
305        statement.validate()?;
306        let operation = statement.operation().to_owned();
307        let call = self.observer.start_child_call(
308            parent,
309            CallKind::Database,
310            "database",
311            "database",
312            OperationId::from(operation),
313        );
314        let result = async {
315            let mut stream = statement.query().fetch(&self.pool);
316            let mut rows = Vec::new();
317            let mut result_bytes = 0_usize;
318            while let Some(row) = stream.try_next().await.map_err(map_operation_error)? {
319                if rows.len() == MAX_QUERY_ROWS {
320                    return Err(result_limit_exceeded());
321                }
322                result_bytes = result_bytes.saturating_add(row_payload_bytes(&row)?);
323                if result_bytes > MAX_RESULT_BYTES {
324                    return Err(result_limit_exceeded());
325                }
326                rows.push(DbRow(row));
327            }
328            Ok(rows)
329        }
330        .await;
331        finish_call(call, &result);
332        result
333    }
334
335    pub async fn query_optional(
336        &self,
337        parent: &CallContext,
338        statement: Statement,
339    ) -> Result<Option<DbRow>> {
340        if self.physical_plans.enabled() {
341            return Err(name_mapping_bypass());
342        }
343        statement.validate()?;
344        let operation = statement.operation().to_owned();
345        let call = self.observer.start_child_call(
346            parent,
347            CallKind::Database,
348            "database",
349            "database",
350            OperationId::from(operation),
351        );
352        let result = statement
353            .query()
354            .fetch_optional(&self.pool)
355            .await
356            .map_err(map_operation_error)
357            .and_then(|row| {
358                row.map(|row| {
359                    row_payload_bytes(&row)?;
360                    Ok(DbRow(row))
361                })
362                .transpose()
363            });
364        finish_call(call, &result);
365        result
366    }
367
368    pub async fn write(&self, parent: &CallContext, statement: Statement) -> Result<WriteResult> {
369        if self.physical_plans.enabled() {
370            return Err(name_mapping_bypass());
371        }
372        statement.validate()?;
373        let operation = statement.operation().to_owned();
374        let call = self.observer.start_child_call(
375            parent,
376            CallKind::Database,
377            "database",
378            "database",
379            OperationId::from(operation),
380        );
381        let result = statement
382            .query()
383            .execute(&self.pool)
384            .await
385            .map(|result| WriteResult::new(result.rows_affected(), result.last_insert_id()))
386            .map_err(map_operation_error);
387        finish_call(call, &result);
388        result
389    }
390
391    /// Runs one explicit, single-level transaction.
392    ///
393    /// The callback returns a boxed async block because that is the minimal
394    /// Rust API that safely ties all operations to the borrowed transaction.
395    pub async fn transaction<T, F>(
396        &self,
397        parent: &CallContext,
398        operation: impl Into<OperationId>,
399        work: F,
400    ) -> Result<T>
401    where
402        T: Send,
403        F: for<'a> FnOnce(&'a mut Transaction) -> TransactionFuture<'a, T> + Send,
404    {
405        if self.physical_plans.enabled() {
406            return Err(name_mapping_bypass());
407        }
408        let operation = operation.into();
409        crate::statement::validate_operation(operation.as_str())?;
410        let call = self.observer.start_child_call(
411            parent,
412            CallKind::Transaction,
413            "database",
414            "database",
415            operation,
416        );
417        let cleanup = match self.cleanup.transaction_sender() {
418            Some(cleanup) => cleanup,
419            None => {
420                let error = transaction_begin_failed();
421                call.fail(&error);
422                return Err(error);
423            }
424        };
425        let begin = self.observer.start_child_call(
426            call.context(),
427            CallKind::Transaction,
428            "database",
429            "database",
430            "begin",
431        );
432        let raw = match self.pool.begin().await {
433            Ok(transaction) => {
434                begin.succeed();
435                transaction
436            }
437            Err(_) => {
438                let error = transaction_begin_failed();
439                begin.fail(&error);
440                call.fail(&error);
441                return Err(error);
442            }
443        };
444        let mut transaction = Transaction::new(raw, self.observer.clone(), call, cleanup);
445        match work(&mut transaction).await {
446            Ok(value) => transaction.commit().await.map(|()| value),
447            Err(work_error) => Err(transaction.rollback(work_error).await),
448        }
449    }
450
451    pub(crate) async fn close(&self) -> Result<()> {
452        let cleanup = self.cleanup.shutdown().await;
453        self.pool.close().await;
454        cleanup
455    }
456
457    #[cfg(test)]
458    pub(crate) fn connect_lazy(config: DatabaseConfig, observer: Observer) -> Result<Self> {
459        let physical_plans = freeze(config.name_mappings.clone())
460            .map_err(|_| invalid_config("database name mapping is invalid"))?;
461        let options = config.options()?;
462        let pool = MySqlPoolOptions::new()
463            .max_connections(config.max_connections)
464            .acquire_timeout(config.acquire_timeout)
465            .connect_lazy_with(options);
466        Ok(Self {
467            pool,
468            observer,
469            cleanup: CleanupCoordinator::start(),
470            physical_plans: Arc::new(physical_plans),
471        })
472    }
473}
474
475impl ComponentLifecycle for Database {
476    fn name(&self) -> &'static str {
477        "database"
478    }
479    fn start(&self) -> LifecycleFuture<'_> {
480        Box::pin(async { Ok(()) })
481    }
482    fn shutdown(&self) -> LifecycleFuture<'_> {
483        Box::pin(async move { self.close().await })
484    }
485}
486
487fn finish_call<T>(call: saddle_observability::ActiveCall, result: &Result<T>) {
488    match result {
489        Ok(_) => call.succeed(),
490        Err(error) => call.fail(error),
491    }
492}
493
494#[cfg(test)]
495mod tests {
496    use saddle_core::{ApplicationId, ErrorKind, ModuleId, ServiceId, SpanId, TraceId};
497    use saddle_observability::ObserverConfig;
498    use serde_json::Value;
499    use std::{
500        io,
501        sync::{Arc, Mutex},
502    };
503
504    use super::*;
505    use crate::SaddleError;
506
507    struct MappedTable;
508
509    impl crate::StaticLogicalTable for MappedTable {
510        const TABLE: &'static str = "逻辑表";
511        const COLUMNS: &'static [&'static str] = &["逻辑列"];
512    }
513
514    #[derive(Clone, Default)]
515    struct Capture(Arc<Mutex<Vec<u8>>>);
516    impl io::Write for Capture {
517        fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
518            self.0.lock().unwrap().extend_from_slice(bytes);
519            Ok(bytes.len())
520        }
521        fn flush(&mut self) -> io::Result<()> {
522            Ok(())
523        }
524    }
525    fn context() -> CallContext {
526        CallContext::new(
527            ApplicationId::from("shop"),
528            ModuleId::from("orders"),
529            ServiceId::from("orders"),
530            OperationId::from("create"),
531            TraceId::from_u128(1),
532            SpanId::from_u64(2),
533        )
534    }
535
536    #[test]
537    fn configuration_rejects_invalid_bounds_without_exposing_url() {
538        let observer = Observer::with_writer(ObserverConfig::default(), io::sink()).unwrap();
539        let error = Database::connect_lazy(
540            DatabaseConfig::new("mysql://secret@localhost/db").max_connections(0),
541            observer,
542        )
543        .err()
544        .unwrap();
545        assert_eq!(error.code(), "db.invalid_config");
546        assert!(!error.to_string().contains("secret"));
547    }
548
549    #[test]
550    fn unified_startup_injection_resolves_secret_and_relative_mapping_directory() {
551        let directory =
552            std::env::temp_dir().join(format!("saddle-db-alpha10-config-{}", std::process::id()));
553        let mappings = directory.join("mappings");
554        let _ = std::fs::remove_dir_all(&directory);
555        std::fs::create_dir_all(&mappings).unwrap();
556        let executable = std::env::current_exe().unwrap();
557        for mode in ["happy", "missing", "bad-path"] {
558            let mut child = std::process::Command::new(&executable);
559            child
560                .args([
561                    "--ignored",
562                    "--exact",
563                    "database::tests::unified_startup_injection_child",
564                ])
565                .env("SADDLE_ALPHA10_INJECTION_MODE", mode)
566                .env("SADDLE_ALPHA10_CONFIG_ROOT", &directory)
567                .env_remove("SADDLE_ALPHA10_DATABASE_TEST");
568            if mode != "missing" {
569                child.env(
570                    "SADDLE_ALPHA10_DATABASE_TEST",
571                    "mysql://deployment-secret@localhost/database",
572                );
573            }
574            assert!(
575                child.status().unwrap().success(),
576                "child mode {mode} failed"
577            );
578        }
579        std::fs::remove_dir_all(directory).unwrap();
580    }
581
582    #[test]
583    #[ignore = "executed in isolated child processes by the parent test"]
584    fn unified_startup_injection_child() {
585        let root = PathBuf::from(std::env::var_os("SADDLE_ALPHA10_CONFIG_ROOT").unwrap());
586        let mode = std::env::var("SADDLE_ALPHA10_INJECTION_MODE").unwrap();
587        let mapping = if mode == "bad-path" {
588            std::path::Path::new("../mappings")
589        } else {
590            std::path::Path::new("mappings")
591        };
592        let result = DatabaseStartupInjection::load(&root, "SADDLE_ALPHA10_DATABASE_TEST", mapping);
593        match mode.as_str() {
594            "happy" => {
595                let injection = result.unwrap();
596                assert_eq!(injection.mapping_directory, root.join("mappings"));
597                assert!(injection.url.contains("deployment-secret"));
598            }
599            "missing" => assert_eq!(
600                result.err(),
601                Some(DatabaseStartupInjectionError::MissingConnectionSecret)
602            ),
603            "bad-path" => assert_eq!(
604                result.err(),
605                Some(DatabaseStartupInjectionError::InvalidMappingDirectory)
606            ),
607            _ => panic!("unknown child mode"),
608        }
609    }
610
611    #[tokio::test]
612    async fn mapping_enabled_rejects_legacy_raw_statement_before_database_io() {
613        let directory =
614            std::env::temp_dir().join(format!("saddle-db-alpha7-bypass-{}", std::process::id()));
615        let _ = std::fs::remove_dir_all(&directory);
616        std::fs::create_dir(&directory).unwrap();
617        std::fs::write(
618            directory.join("table.json"),
619            r#"{"table":{"from":"逻辑表","to":"physical_table"},"columns":[{"from":"逻辑列","to":"physical_column"}]}"#,
620        )
621        .unwrap();
622        let observer = Observer::with_writer(ObserverConfig::default(), io::sink()).unwrap();
623        let database = Database::connect_lazy(
624            DatabaseConfig::new("mysql://localhost/unused")
625                .name_mapping_directory(&directory)
626                .register_logical_table::<MappedTable>(),
627            observer,
628        )
629        .unwrap();
630        let error = database
631            .write(
632                &context(),
633                Statement::new("legacy.write", "SELECT 1").unwrap(),
634            )
635            .await
636            .unwrap_err();
637        assert_eq!(error.code(), "db.name_mapping_required");
638        database.shutdown().await.unwrap();
639        std::fs::remove_dir_all(directory).unwrap();
640    }
641
642    #[tokio::test]
643    async fn closed_pool_query_has_stable_error_and_trace_record() {
644        let capture = Capture::default();
645        let observer = Observer::with_writer(ObserverConfig::default(), capture.clone()).unwrap();
646        let database = Database::connect_lazy(
647            DatabaseConfig::new("mysql://localhost/db"),
648            observer.clone(),
649        )
650        .unwrap();
651        database.close().await.unwrap();
652        let error = match database
653            .query_all(
654                &context(),
655                Statement::new("orders.list", "SELECT 1").unwrap(),
656            )
657            .await
658        {
659            Ok(_) => panic!("closed pool query unexpectedly succeeded"),
660            Err(error) => error,
661        };
662        assert_eq!(error.kind(), ErrorKind::Unavailable);
663        assert_eq!(error.code(), "db.connection_unavailable");
664        observer.flush().await.unwrap();
665        let output = String::from_utf8(capture.0.lock().unwrap().clone()).unwrap();
666        let records: Vec<Value> = output
667            .lines()
668            .map(|line| serde_json::from_str(line).unwrap())
669            .collect();
670        assert_eq!(records[0]["call_kind"], "database");
671        assert_eq!(records[1]["error_code"], "db.connection_unavailable");
672        assert_eq!(records[0]["trace_id"], context().trace_id().to_string());
673        assert!(!output.contains("SELECT 1"));
674    }
675
676    #[tokio::test]
677    async fn transaction_begin_failure_records_phase_and_stable_error() {
678        let capture = Capture::default();
679        let observer = Observer::with_writer(ObserverConfig::default(), capture.clone()).unwrap();
680        let database = Database::connect_lazy(
681            DatabaseConfig::new("mysql://localhost/db"),
682            observer.clone(),
683        )
684        .unwrap();
685        database.pool.close().await;
686        let error = database
687            .transaction(&context(), "orders.create", |_transaction| {
688                Box::pin(async { Ok::<_, SaddleError>(()) })
689            })
690            .await
691            .unwrap_err();
692        assert_eq!(error.code(), "db.transaction_begin_failed");
693        observer.flush().await.unwrap();
694        let output = String::from_utf8(capture.0.lock().unwrap().clone()).unwrap();
695        let records: Vec<Value> = output
696            .lines()
697            .map(|line| serde_json::from_str(line).unwrap())
698            .collect();
699        assert_eq!(records.len(), 4);
700        assert_eq!(records[0]["operation"], "orders.create");
701        assert_eq!(records[1]["operation"], "begin");
702        assert_eq!(records[2]["error_code"], "db.transaction_begin_failed");
703        assert_eq!(records[3]["error_code"], "db.transaction_begin_failed");
704        assert!(
705            records
706                .iter()
707                .all(|record| record["trace_id"] == context().trace_id().to_string())
708        );
709    }
710
711    #[allow(dead_code)]
712    async fn transaction_usage_compiles(database: &Database, context: &CallContext) -> Result<()> {
713        database
714            .transaction(context, "orders.create", |transaction| {
715                Box::pin(async move {
716                    transaction
717                        .write(
718                            Statement::new("orders.insert", "INSERT INTO orders(id) VALUES (?)")?
719                                .bind(1_u64)?,
720                        )
721                        .await?;
722                    transaction
723                        .query_optional(
724                            Statement::new("orders.find", "SELECT id FROM orders WHERE id = ?")?
725                                .bind(1_u64)?,
726                        )
727                        .await?;
728                    Ok(())
729                })
730            })
731            .await
732    }
733}
saddle_db/database.rs

saddle_db/
database.rs
