forge
ExploreLog in

Tahinli/forge

public
0 watching
★ 0
4.6 KiB
1//! Concurrency spike for the store read pool: one turso `Database` with several connections.
2//! Checks read-after-commit visibility, isolation from an open writer transaction, and that
3//! concurrent readers next to a busy writer neither error nor report "database is locked".
4
5use std::sync::Arc;
6use std::time::Duration;
7
8use turso::{Builder, Connection, Database};
9
10const BUSY_TIMEOUT: Duration = Duration::from_millis(5000);
11
12async fn open_db(dir: &tempfile::TempDir) -> Database {
13 let path = dir.path().join("spike.db");
14 let db = Builder::new_local(path.to_str().expect("utf-8 path"))
15 .build()
16 .await
17 .expect("build database");
18 let writer = db.connect().expect("writer connect");
19 let mut rows = writer
20 .query("PRAGMA journal_mode = WAL", ())
21 .await
22 .expect("wal");
23 while rows.next().await.expect("wal row").is_some() {}
24 writer
25 .execute(
26 "CREATE TABLE IF NOT EXISTS t (id INTEGER PRIMARY KEY, v INTEGER NOT NULL)",
27 (),
28 )
29 .await
30 .expect("create table");
31 db
32}
33
34/// A connection configured the way the store's readers will be: busy timeout and query_only.
35async fn reader_conn(db: &Database) -> Connection {
36 let conn = db.connect().expect("reader connect");
37 conn.busy_timeout(BUSY_TIMEOUT).expect("busy timeout");
38 let mut rows = conn
39 .query("PRAGMA query_only = 1", ())
40 .await
41 .expect("query_only");
42 while rows.next().await.expect("query_only row").is_some() {}
43 conn
44}
45
46fn writer_conn(db: &Database) -> Connection {
47 let conn = db.connect().expect("writer connect");
48 conn.busy_timeout(BUSY_TIMEOUT).expect("busy timeout");
49 conn
50}
51
52async fn count(conn: &Connection) -> i64 {
53 let mut rows = conn
54 .query("SELECT COUNT(*) FROM t", ())
55 .await
56 .expect("count query");
57 let row = rows.next().await.expect("count row").expect("one row");
58 row.get::<i64>(0).expect("count value")
59}
60
61#[tokio::test]
62async fn reader_sees_committed_rows_immediately() {
63 let dir = tempfile::tempdir().unwrap();
64 let db = open_db(&dir).await;
65 let writer = writer_conn(&db);
66 let reader = reader_conn(&db).await;
67
68 assert_eq!(count(&reader).await, 0);
69 writer
70 .execute("INSERT INTO t (v) VALUES (1)", ())
71 .await
72 .expect("insert");
73 assert_eq!(
74 count(&reader).await,
75 1,
76 "reader must see the autocommitted row"
77 );
78}
79
80#[tokio::test]
81async fn reader_does_not_see_uncommitted_rows_and_does_not_block() {
82 let dir = tempfile::tempdir().unwrap();
83 let db = open_db(&dir).await;
84 let writer = writer_conn(&db);
85 let reader = reader_conn(&db).await;
86
87 writer.execute("BEGIN", ()).await.expect("begin");
88 writer
89 .execute("INSERT INTO t (v) VALUES (2)", ())
90 .await
91 .expect("insert in tx");
92 writer
93 .execute("INSERT INTO t (v) VALUES (3)", ())
94 .await
95 .expect("insert in tx");
96
97 // The writer still holds its transaction open here.
98 let seen = tokio::time::timeout(BUSY_TIMEOUT, count(&reader))
99 .await
100 .expect("reader must not block on an open writer transaction");
101 assert_eq!(seen, 0, "reader must not see uncommitted rows");
102
103 writer.execute("COMMIT", ()).await.expect("commit");
104 assert_eq!(count(&reader).await, 2, "reader must see rows after commit");
105}
106
107#[tokio::test]
108async fn eight_readers_with_busy_writer_all_succeed() {
109 let dir = tempfile::tempdir().unwrap();
110 let db = Arc::new(open_db(&dir).await);
111 let writer = writer_conn(&db);
112
113 let mut readers = Vec::new();
114 for _ in 0..8 {
115 let db = Arc::clone(&db);
116 readers.push(tokio::spawn(async move {
117 let conn = reader_conn(&db).await;
118 let mut last = 0;
119 for _ in 0..200 {
120 let n = count(&conn).await;
121 assert!(n >= last, "reader saw count go backwards: {last} -> {n}");
122 last = n;
123 }
124 last
125 }));
126 }
127
128 for i in 0..200 {
129 writer
130 .execute("INSERT INTO t (v) VALUES (?1)", (i as i64,))
131 .await
132 .expect("writer insert must not fail");
133 }
134
135 for handle in readers {
136 handle.await.expect("reader task panicked");
137 }
138 assert_eq!(count(&reader_conn(&db).await).await, 200);
139}
140
141#[tokio::test]
142async fn reader_rejects_writes_when_query_only() {
143 let dir = tempfile::tempdir().unwrap();
144 let db = open_db(&dir).await;
145 let reader = reader_conn(&db).await;
146 let result = reader.execute("INSERT INTO t (v) VALUES (9)", ()).await;
147 assert!(result.is_err(), "query_only reader must refuse writes");
148 assert_eq!(count(&reader).await, 0);
149}