miden_node_db/sqlite/
tx.rs

1//! Read/write transaction wrappers and the [`Row`] accessor.
2//!
3//! [`ReadTx`] and [`WriteTx`] are the only handles callers ever touch; the underlying
4//! `rusqlite::Connection` is private. They borrow a connection on which a transaction has already
5//! been opened (by the pool's `read`/`write` or by a held transaction handle) — they do not begin
6//! or end the transaction themselves.
7//!
8//! A write transaction is a read transaction with the extra ability to mutate: [`WriteTx`] wraps a
9//! [`ReadTx`] and derefs to it, so it inherits [`query`](ReadTx::query) and only adds
10//! [`execute`](WriteTx::execute). A function that receives `&ReadTx` therefore cannot compile a
11//! mutation. Both prepare statements with `prepare_cached`, so prepared statements are always
12//! cached.
13
14use std::ops::Deref;
15
16use rusqlite::Connection;
17
18use crate::DatabaseError;
19use crate::sqlite::codec::{DbValue, DbValueRef, FromSqlValue, ToSqlValue};
20
21// ROW
22// =================================================================================================
23
24/// A single result row. Wraps `rusqlite::Row` so callers read columns through the codec via
25/// [`Row::get`] without naming `rusqlite`.
26pub struct Row<'a>(&'a rusqlite::Row<'a>);
27
28impl<'a> Row<'a> {
29    fn new(row: &'a rusqlite::Row<'a>) -> Self {
30        Self(row)
31    }
32
33    /// Reads column `idx` (zero-based) decoded through [`FromSqlValue`], e.g.
34    /// `row.get::<BlockHeader>(0)`.
35    pub fn get<T: FromSqlValue>(&self, idx: usize) -> Result<T, DatabaseError> {
36        let value = self.0.get_ref(idx)?;
37        T::from_sql_value(DbValueRef::new(value))
38    }
39}
40
41// TRANSACTION WRAPPERS
42// =================================================================================================
43
44/// A read-only transaction. Borrows a connection on which a `DEFERRED` transaction is open; the
45/// transaction is never committed (changes roll back when it ends).
46pub struct ReadTx<'t>(&'t Connection);
47
48impl<'t> ReadTx<'t> {
49    pub(crate) fn new(conn: &'t Connection) -> Self {
50        Self(conn)
51    }
52
53    /// Runs a query and maps every row, collecting the results.
54    ///
55    /// This is the single read primitive: a caller expecting at most one row takes
56    /// `.into_iter().next()`, and `SELECT EXISTS(...)` / `SELECT COUNT(*)` map the single row's
57    /// first column.
58    pub fn query<T>(
59        &self,
60        sql: &'static str,
61        params: &[&dyn ToSqlValue],
62        mut map: impl FnMut(&Row<'_>) -> Result<T, DatabaseError>,
63    ) -> Result<Vec<T>, DatabaseError> {
64        debug_assert_no_dynamic_in(sql);
65        let values = to_values(params);
66        let mut stmt = self.0.prepare_cached(sql)?;
67        let mut rows = stmt.query(rusqlite::params_from_iter(values))?;
68        let mut out = Vec::new();
69        while let Some(row) = rows.next()? {
70            out.push(map(&Row::new(row))?);
71        }
72        Ok(out)
73    }
74}
75
76/// A read-write transaction. Borrows a connection on which an `IMMEDIATE` transaction is open; the
77/// transaction is committed by the owner when the work returns `Ok`. Derefs to [`ReadTx`] for all
78/// read queries and adds [`execute`](Self::execute).
79pub struct WriteTx<'t>(ReadTx<'t>);
80
81impl<'t> WriteTx<'t> {
82    pub(crate) fn new(conn: &'t Connection) -> Self {
83        Self(ReadTx::new(conn))
84    }
85
86    /// Executes an `INSERT`/`UPDATE`/`DELETE`/`REPLACE` and returns the affected row count.
87    pub fn execute(
88        &self,
89        sql: &'static str,
90        params: &[&dyn ToSqlValue],
91    ) -> Result<usize, DatabaseError> {
92        debug_assert_no_dynamic_in(sql);
93        let values = to_values(params);
94        let mut stmt = self.0.0.prepare_cached(sql)?;
95        Ok(stmt.execute(rusqlite::params_from_iter(values))?)
96    }
97}
98
99impl<'t> Deref for WriteTx<'t> {
100    type Target = ReadTx<'t>;
101
102    fn deref(&self) -> &Self::Target {
103        &self.0
104    }
105}
106
107// SHARED HELPERS
108// =================================================================================================
109
110fn to_values(params: &[&dyn ToSqlValue]) -> Vec<DbValue> {
111    params.iter().map(ToSqlValue::to_sql_value).collect()
112}
113
114fn debug_assert_no_dynamic_in(sql: &str) {
115    debug_assert!(
116        !(sql.contains(" IN (?") || sql.contains(" IN (:")),
117        "use in_list() instead of a variable-length `IN (?, ...)` placeholder list to keep the \
118         statement cacheable: {sql}"
119    );
120}
121
122#[cfg(test)]
123mod tests {
124    use rusqlite::Connection;
125
126    use super::*;
127    use crate::sqlite::InList;
128
129    /// Opens a connection to a fresh, file-backed database in a temporary directory.
130    fn temp_db() -> (tempfile::TempDir, Connection) {
131        let dir = tempfile::tempdir().expect("create temp dir");
132        let conn = Connection::open(dir.path().join("test.sqlite3")).expect("open db");
133        // `rarray()` is provided by rusqlite's `array` extension, which must be loaded per
134        // connection (the pool does this in `configure_connection`).
135        rusqlite::vtab::array::load_module(&conn).expect("load array module");
136        conn.execute_batch(
137            "CREATE TABLE items (id INTEGER PRIMARY KEY, payload BLOB, label TEXT);",
138        )
139        .expect("create table");
140        (dir, conn)
141    }
142
143    #[test]
144    fn write_then_read_roundtrips_through_the_codec() {
145        let (_dir, mut conn) = temp_db();
146        let tx = conn.transaction().unwrap();
147        let w = WriteTx::new(&tx);
148
149        let payload = vec![1u8, 2, 3];
150        let inserted = w
151            .execute(
152                "INSERT INTO items (id, payload, label) VALUES (?1, ?2, ?3)",
153                &[&1i64, &payload, &"hello".to_string()],
154            )
155            .unwrap();
156        assert_eq!(inserted, 1);
157
158        let got: (i64, Vec<u8>, String) = w
159            .query("SELECT id, payload, label FROM items WHERE id = ?1", &[&1i64], |row| {
160                Ok((row.get::<i64>(0)?, row.get::<Vec<u8>>(1)?, row.get::<String>(2)?))
161            })
162            .unwrap()
163            .into_iter()
164            .next()
165            .unwrap();
166        assert_eq!(got, (1, vec![1, 2, 3], "hello".to_string()));
167    }
168
169    #[test]
170    fn null_column_reads_as_none() {
171        let (_dir, mut conn) = temp_db();
172        let tx = conn.transaction().unwrap();
173        let w = WriteTx::new(&tx);
174
175        w.execute("INSERT INTO items (id, payload) VALUES (?1, NULL)", &[&1i64])
176            .unwrap();
177        let payload: Option<Vec<u8>> = w
178            .query("SELECT payload FROM items WHERE id = ?1", &[&1i64], |row| {
179                row.get::<Option<Vec<u8>>>(0)
180            })
181            .unwrap()
182            .into_iter()
183            .next()
184            .unwrap();
185        assert_eq!(payload, None);
186    }
187
188    #[test]
189    fn query_returns_empty_for_missing_row() {
190        let (_dir, mut conn) = temp_db();
191        let tx = conn.transaction().unwrap();
192        let r = ReadTx::new(&tx);
193
194        let got = r
195            .query("SELECT id FROM items WHERE id = ?1", &[&404i64], |row| row.get::<i64>(0))
196            .unwrap();
197        assert!(got.is_empty());
198    }
199
200    // Regression guard for the cacheable IN-list idiom: the rarray form must run through `query`
201    // without tripping `debug_assert_no_dynamic_in` (tests run with debug assertions on).
202    #[test]
203    fn in_list_i64_rarray_runs_and_matches() {
204        let (_dir, mut conn) = temp_db();
205        let tx = conn.transaction().unwrap();
206        let w = WriteTx::new(&tx);
207        for id in [1i64, 2, 3, 4] {
208            w.execute("INSERT INTO items (id) VALUES (?1)", &[&id]).unwrap();
209        }
210
211        let wanted = InList::from_i64s([1, 3]);
212        let mut ids = w
213            .query(
214                "SELECT id FROM items WHERE id IN (SELECT value FROM rarray(?1))",
215                &[&wanted],
216                |row| row.get::<i64>(0),
217            )
218            .unwrap();
219        ids.sort_unstable();
220        assert_eq!(ids, vec![1, 3]);
221    }
222
223    #[test]
224    fn in_list_blob_matches_blob_column() {
225        let (_dir, mut conn) = temp_db();
226        let tx = conn.transaction().unwrap();
227        let w = WriteTx::new(&tx);
228        let a = vec![0xAAu8, 0xBB];
229        let b = vec![0x01u8];
230        w.execute("INSERT INTO items (id, payload) VALUES (1, ?1)", &[&a]).unwrap();
231        w.execute("INSERT INTO items (id, payload) VALUES (2, ?1)", &[&b]).unwrap();
232
233        let wanted = InList::from_blobs([a.as_slice()]);
234        let ids = w
235            .query(
236                "SELECT id FROM items WHERE payload IN (SELECT value FROM rarray(?1))",
237                &[&wanted],
238                |row| row.get::<i64>(0),
239            )
240            .unwrap();
241        assert_eq!(ids, vec![1]);
242    }
243}
miden_node_db/sqlite/tx.rs

miden_node_db/sqlite/
tx.rs
