1 parent
95c0587c1e
commit
5b22add1f3
21 files changed
+293
-89
No files matched your search
+75
-12
@@ -9,7 +9,7 @@ use rusqlite::OptionalExtension as _;
|
||||
use rusqlite::types::{ToSql, ToSqlOutput, Value};
|
||||
use sha2::{Digest, Sha256};
|
||||
use sqlx::pool::PoolConnection;
|
||||
use sqlx::{MySql, MySqlPool, PgPool, Postgres, Row as SqlxRow};
|
||||
use sqlx::{Executor as _, MySql, MySqlPool, PgPool, Postgres, Row as SqlxRow};
|
||||
use tokio::runtime::{Builder, Handle};
|
||||
use tokio::sync::Mutex as AsyncMutex;
|
||||
|
||||
@@ -180,14 +180,6 @@ fn global_schema(schema: &str) -> Result<String, String> {
|
||||
schema_section(schema, "-- RUSTBOX:GLOBAL-BEGIN", "-- RUSTBOX:GLOBAL-END")
|
||||
}
|
||||
|
||||
pub(crate) fn global_schema_for_backend(backend: DatabaseBackend) -> Result<String, String> {
|
||||
match backend {
|
||||
DatabaseBackend::Sqlite => global_schema(SQLITE_SCHEMA),
|
||||
DatabaseBackend::Mysql => global_schema(MYSQL_SCHEMA),
|
||||
DatabaseBackend::Pgsql => global_schema(PGSQL_SCHEMA),
|
||||
}
|
||||
}
|
||||
|
||||
fn tenant_schema(schema: &str, prefix: &str) -> Result<String, String> {
|
||||
Ok(
|
||||
schema_section(schema, "-- RUSTBOX:TENANT-BEGIN", "-- RUSTBOX:TENANT-END")?
|
||||
@@ -236,7 +228,22 @@ pub async fn configure(config: &AppConfig) -> Result<(), String> {
|
||||
.map_err(|e| format!("failed to configure SQLite database: {e}"))?;
|
||||
conn.execute_batch(&global_schema(SQLITE_SCHEMA)?)
|
||||
.map_err(|e| format!("SQLite global schema initialization failed: {e}"))?;
|
||||
ConfiguredDatabase::Sqlite { path }
|
||||
|
||||
// Unit tests create many independent temporary data directories in one
|
||||
// process. Pinning the first SQLite path in the process-wide OnceLock
|
||||
// makes later tests reuse a TempDir that may already have been deleted,
|
||||
// causing a cascade of "unable to open database file" failures.
|
||||
// Initialize the requested test database, but let open_user_db() resolve
|
||||
// each test's own data_dir instead of publishing this path globally.
|
||||
#[cfg(test)]
|
||||
{
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
#[cfg(not(test))]
|
||||
{
|
||||
ConfiguredDatabase::Sqlite { path }
|
||||
}
|
||||
}
|
||||
DatabaseBackend::Mysql => {
|
||||
let url = config
|
||||
@@ -280,10 +287,29 @@ pub async fn configure(config: &AppConfig) -> Result<(), String> {
|
||||
let schema = global_schema(PGSQL_SCHEMA)?;
|
||||
let init_pool = pool.clone();
|
||||
runtime_wait(&runtime, async move {
|
||||
sqlx::raw_sql(&schema)
|
||||
.execute(&init_pool)
|
||||
// PostgreSQL emits NOTICE for every CREATE ... IF NOT EXISTS that
|
||||
// already exists. Keep expected idempotent startup DDL quiet without
|
||||
// changing application-wide logging or server settings.
|
||||
let mut tx = init_pool
|
||||
.begin()
|
||||
.await
|
||||
.map_err(|e| format!("PostgreSQL global schema transaction failed: {e}"))?;
|
||||
sqlx::query("SET LOCAL client_min_messages TO WARNING")
|
||||
.execute(&mut *tx)
|
||||
.await
|
||||
.map_err(|e| format!("failed to quiet PostgreSQL schema notices: {e}"))?;
|
||||
// `raw_sql(...).execute(&mut PgConnection)` can trigger SQLx's
|
||||
// "Executor is not general enough" lifetime error when this
|
||||
// future is spawned onto the dedicated database runtime. Execute
|
||||
// the unprepared multi-statement DDL directly through `Executor`
|
||||
// instead; this is SQLx's supported path for raw DDL batches.
|
||||
(&mut *tx)
|
||||
.execute(schema.as_str())
|
||||
.await
|
||||
.map_err(|e| format!("PostgreSQL global schema initialization failed: {e}"))?;
|
||||
tx.commit()
|
||||
.await
|
||||
.map_err(|e| format!("PostgreSQL global schema commit failed: {e}"))?;
|
||||
Ok(())
|
||||
})?;
|
||||
ConfiguredDatabase::Pgsql { pool, runtime }
|
||||
@@ -2287,6 +2313,31 @@ pub fn open_test_db() -> Connection {
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[tokio::test]
|
||||
async fn sqlite_test_configuration_does_not_pin_deleted_tempdir() {
|
||||
let first = tempfile::tempdir().expect("failed to create first temp dir");
|
||||
let config = AppConfig {
|
||||
database_backend: "sqlite".to_string(),
|
||||
database_url: None,
|
||||
data_dir: first.path().to_string_lossy().into_owned(),
|
||||
..Default::default()
|
||||
};
|
||||
|
||||
configure(&config)
|
||||
.await
|
||||
.expect("failed to initialize first SQLite test database");
|
||||
assert!(first.path().join("rustbox.sqlite").is_file());
|
||||
drop(first);
|
||||
|
||||
let second = tempfile::tempdir().expect("failed to create second temp dir");
|
||||
let second_dir = second.path().to_string_lossy().into_owned();
|
||||
let connection = open_user_db(&second_dir, "second-test-account")
|
||||
.expect("second test must not reuse the deleted first SQLite path");
|
||||
|
||||
assert_eq!(connection.backend(), DatabaseBackend::Sqlite);
|
||||
assert!(second.path().join("rustbox.sqlite").is_file());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn table_prefix_is_short_and_deterministic() {
|
||||
let hash = "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef";
|
||||
@@ -2322,6 +2373,18 @@ mod tests {
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn tenant_rewriter_prefixes_header_sync_target_qualifier() {
|
||||
let sql = prefix_tenant_tables(
|
||||
"INSERT INTO user_meta (key, value) VALUES ('header_sync_generation', '1') ON CONFLICT(key) DO UPDATE SET value = CAST(user_meta.value AS INTEGER) + 1",
|
||||
"u_test__",
|
||||
);
|
||||
assert_eq!(
|
||||
sql,
|
||||
"INSERT INTO u_test__user_meta (key, value) VALUES ('header_sync_generation', '1') ON CONFLICT(key) DO UPDATE SET value = CAST(u_test__user_meta.value AS INTEGER) + 1"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_schema_contains_sessions_and_user_tables() {
|
||||
let conn = open_test_db();
|
||||
|
||||
@@ -243,7 +243,7 @@ pub mod db_header_sync {
|
||||
pub const Q_001: &str = "SELECT value FROM user_meta WHERE key = 'header_sync_generation'";
|
||||
pub const Q_002: &str =
|
||||
"INSERT INTO user_meta (key, value) VALUES ('header_sync_generation', '1')
|
||||
ON CONFLICT(key) DO UPDATE SET value = CAST(value AS INTEGER) + 1";
|
||||
ON CONFLICT(key) DO UPDATE SET value = CAST(user_meta.value AS INTEGER) + 1";
|
||||
}
|
||||
|
||||
pub mod db_identities {
|
||||
|
||||
@@ -1114,10 +1114,7 @@ async fn process_search_outbox(
|
||||
Err(error) => {
|
||||
if is_opensearch_backpressure_error(&error) {
|
||||
crate::db::search_outbox::defer_transient_failure(
|
||||
&conn,
|
||||
&event,
|
||||
&error,
|
||||
30,
|
||||
&conn, &event, &error, 30,
|
||||
)?;
|
||||
return Err(format!(
|
||||
"OpenSearch temporarily rejected indexing for tenant {user_hash}: {error}"
|
||||
|
||||
@@ -6,7 +6,6 @@ use sqlx::{FromRow, MySqlPool, PgPool};
|
||||
|
||||
use super::{PersistedSession, SessionPersistence};
|
||||
use crate::config::AppConfig;
|
||||
use crate::db::pool::{DatabaseBackend, global_schema_for_backend};
|
||||
use crate::queries;
|
||||
|
||||
#[derive(Clone)]
|
||||
@@ -149,20 +148,15 @@ pub struct DatabaseSessionStore {
|
||||
|
||||
impl DatabaseSessionStore {
|
||||
pub async fn connect(config: &AppConfig) -> Result<Self, String> {
|
||||
// Schema ownership is centralized in db::pool::configure(). Calling it here
|
||||
// keeps direct/test construction safe, while the normal application startup
|
||||
// hits the OnceLock fast path. Do not run schema.sql again from this store:
|
||||
// that used to duplicate DDL for SQLite/MySQL/PostgreSQL and produced repeated
|
||||
// PostgreSQL "relation already exists, skipping" NOTICE messages.
|
||||
crate::db::pool::configure(config).await?;
|
||||
|
||||
let pool = match config.database_backend.as_str() {
|
||||
"sqlite" => {
|
||||
let path = sqlite_path(config)?;
|
||||
let migration_path = path.clone();
|
||||
let schema = global_schema_for_backend(DatabaseBackend::Sqlite)?;
|
||||
sqlite_task(migration_path, move |conn| {
|
||||
conn.execute_batch(&schema).map_err(|e| {
|
||||
format!("SQLite database schema initialization failed: {e}")
|
||||
})?;
|
||||
Ok(())
|
||||
})
|
||||
.await?;
|
||||
Pool::Sqlite(path)
|
||||
}
|
||||
"sqlite" => Pool::Sqlite(sqlite_path(config)?),
|
||||
"mysql" => {
|
||||
let url = config
|
||||
.database_url
|
||||
@@ -171,11 +165,6 @@ impl DatabaseSessionStore {
|
||||
let pool = MySqlPool::connect(url)
|
||||
.await
|
||||
.map_err(|e| format!("failed to connect MySQL database: {e}"))?;
|
||||
let schema = global_schema_for_backend(DatabaseBackend::Mysql)?;
|
||||
sqlx::raw_sql(&schema)
|
||||
.execute(&pool)
|
||||
.await
|
||||
.map_err(|e| format!("MySQL database schema initialization failed: {e}"))?;
|
||||
Pool::Mysql(pool)
|
||||
}
|
||||
"pgsql" => {
|
||||
@@ -186,10 +175,6 @@ impl DatabaseSessionStore {
|
||||
let pool = PgPool::connect(url)
|
||||
.await
|
||||
.map_err(|e| format!("failed to connect PostgreSQL database: {e}"))?;
|
||||
let schema = global_schema_for_backend(DatabaseBackend::Pgsql)?;
|
||||
sqlx::raw_sql(&schema).execute(&pool).await.map_err(|e| {
|
||||
format!("PostgreSQL database schema initialization failed: {e}")
|
||||
})?;
|
||||
Pool::Pgsql(pool)
|
||||
}
|
||||
other => return Err(format!("unsupported database backend: {other}")),
|
||||
|
||||
Reference in new issue
Block a user