diff --git a/server/src/error.rs b/server/src/error.rs index 32abc18..e4285e7 100644 --- a/server/src/error.rs +++ b/server/src/error.rs @@ -25,6 +25,10 @@ pub enum Error { Database(#[from] sqlx::Error), #[error("database migration error: {0}")] Migration(#[from] sqlx::migrate::MigrateError), + #[error( + "database migration stage '{stage}' timed out after {timeout_seconds} seconds; close other Cursor BYOK processes and try again" + )] + MigrationTimeout { stage: String, timeout_seconds: u64 }, #[error("http error: {0}")] Http(#[from] reqwest::Error), #[error("protobuf decode error: {0}")] @@ -49,6 +53,7 @@ impl IntoResponse for Error { Self::Store(_) | Self::Database(_) | Self::Migration(_) + | Self::MigrationTimeout { .. } | Self::Encode(_) | Self::Io(_) => StatusCode::INTERNAL_SERVER_ERROR, }; diff --git a/server/src/run/compaction.rs b/server/src/run/compaction.rs index 0f5bc56..c382db6 100644 --- a/server/src/run/compaction.rs +++ b/server/src/run/compaction.rs @@ -6,7 +6,6 @@ use crate::model::{ CanonicalMessage, LlmCallUsageAnchor, PreparedRun, ProjectedMessage, RunAction, }; -const RESERVE_TOKENS: u64 = 10_000; const FALLBACK_CHARS: usize = 12_000; pub(super) const OUTPUT_TOKENS: u64 = 4_096; @@ -41,7 +40,7 @@ pub(super) fn should_compact( let Some(context_window) = prepared.model.context_window_tokens else { return false; }; - if context_window <= RESERVE_TOKENS || messages.len() <= prepared.initial_messages.len() { + if context_window == 0 || messages.len() <= prepared.initial_messages.len() { return false; } let estimated_input = anchor @@ -62,7 +61,7 @@ pub(super) fn should_compact( &serde_json::to_string(&(&prepared.prompt, messages)).unwrap_or_default(), ) }); - estimated_input > context_window.saturating_sub(RESERVE_TOKENS) + estimated_input > context_window } pub(super) fn partition( @@ -109,3 +108,66 @@ fn estimate_serialized_tokens(serialized: &str) -> u64 { }) .div_ceil(1_000) } + +#[cfg(test)] +mod tests { + use super::*; + use crate::model::{ + project_messages, CheckpointId, ConversationId, ModelSpec, Origin, PromptSpec, Role, RunId, + RunKind, + }; + + #[test] + fn automatic_compaction_starts_only_after_the_context_window_is_exceeded() { + let mut model = ModelSpec::new("model"); + model.context_window_tokens = Some(200_000); + let prepared = PreparedRun { + run_id: RunId::new("run"), + cursor_request_id: None, + conversation_id: ConversationId::new("conversation"), + kind: RunKind::Root, + model, + prompt: PromptSpec { + instructions: String::new(), + tools: Vec::new(), + }, + initial_messages: Vec::new(), + action: RunAction::Start, + base_checkpoint_id: CheckpointId(1), + }; + let messages = vec![CanonicalMessage::text( + "user", + Role::User, + Origin::Runtime, + "hello", + )]; + let projected = project_messages(&messages).unwrap(); + let tail_tokens = estimate_serialized_tokens(&serde_json::to_string(&projected).unwrap()); + let anchor = |estimated_input| { + Some(ContextUsageAnchor { + input_tokens: estimated_input - tail_tokens, + message_count: 0, + tool_count: 0, + }) + }; + + assert!(!should_compact( + &prepared, + &messages, + &projected, + anchor(199_999) + )); + assert!(!should_compact( + &prepared, + &messages, + &projected, + anchor(200_000) + )); + assert!(should_compact( + &prepared, + &messages, + &projected, + anchor(200_001) + )); + } +} diff --git a/server/src/run/event.rs b/server/src/run/event.rs index d122a9d..ba6da13 100644 --- a/server/src/run/event.rs +++ b/server/src/run/event.rs @@ -36,6 +36,12 @@ impl From for RunFailure { Error::Http(error) => Self::Provider(error.to_string()), Error::Database(error) => Self::Store(error.to_string()), Error::Migration(error) => Self::Store(error.to_string()), + Error::MigrationTimeout { + stage, + timeout_seconds, + } => Self::Store(format!( + "database migration stage '{stage}' timed out after {timeout_seconds} seconds" + )), Error::Io(error) => Self::Store(error.to_string()), Error::Decode(error) => Self::Protocol(error.to_string()), Error::Encode(error) => Self::Protocol(error.to_string()), diff --git a/server/src/search/cache.rs b/server/src/search/cache.rs new file mode 100644 index 0000000..3222078 --- /dev/null +++ b/server/src/search/cache.rs @@ -0,0 +1,137 @@ +//! Persists fetched web content and serves it from the existing server. +use std::{ + fs, + net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr}, + path::{Path, PathBuf}, + sync::Arc, +}; + +use axum::Router; +use parking_lot::RwLock; +use tower_http::services::ServeDir; +use uuid::Uuid; + +use crate::{config::managed_data_dir, Error, Result}; + +const CACHE_ROUTE: &str = "/web-cache"; + +#[derive(Clone, Default)] +pub struct WebCache { + inner: Option>, +} + +struct WebCacheInner { + directory: PathBuf, + service_addr: RwLock>, +} + +impl WebCache { + pub fn managed() -> Result { + Self::at(managed_data_dir()?.join("cache/web")) + } + + pub fn at(directory: PathBuf) -> Result { + fs::create_dir_all(&directory)?; + Ok(Self { + inner: Some(Arc::new(WebCacheInner { + directory, + service_addr: RwLock::new(None), + })), + }) + } + + pub fn set_service_addr(&self, address: SocketAddr) { + if let Some(inner) = &self.inner { + *inner.service_addr.write() = Some(local_address(address)); + } + } + + pub async fn store(&self, content: &str) -> Result> { + let Some(inner) = &self.inner else { + return Ok(None); + }; + let address = inner.service_addr.read().ok_or_else(|| { + Error::Config("web cache is unavailable before the server starts listening".into()) + })?; + let file_name = format!("{}.txt", Uuid::new_v4()); + let path = inner.directory.join(&file_name); + let bytes = content.as_bytes().to_vec(); + tokio::task::spawn_blocking(move || fs::write(path, bytes)) + .await + .map_err(|error| Error::Store(format!("web cache write task failed: {error}")))??; + Ok(Some(format!("http://{address}{CACHE_ROUTE}/{file_name}"))) + } + + pub fn router(&self) -> Router { + let Some(inner) = &self.inner else { + return Router::new(); + }; + Router::new().nest_service(CACHE_ROUTE, ServeDir::new(inner.directory.clone())) + } + + #[cfg(test)] + fn directory(&self) -> &Path { + &self.inner.as_ref().expect("enabled web cache").directory + } +} + +fn local_address(address: SocketAddr) -> SocketAddr { + match address.ip() { + IpAddr::V4(ip) if ip.is_unspecified() => { + SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), address.port()) + } + IpAddr::V6(ip) if ip.is_unspecified() => { + SocketAddr::new(IpAddr::V6(Ipv6Addr::LOCALHOST), address.port()) + } + _ => address, + } +} + +#[cfg(test)] +mod tests { + use axum::{ + body::{to_bytes, Body}, + http::{Request, StatusCode}, + }; + use tempfile::tempdir; + use tower::ServiceExt; + use uuid::Uuid; + + use super::WebCache; + + #[tokio::test] + async fn stores_uuid_named_content_and_serves_it_from_existing_router() { + let directory = tempdir().unwrap(); + let cache = WebCache::at(directory.path().join("cache/web")).unwrap(); + cache + .set_service_addr("0.0.0.0:4312".parse().unwrap()); + + let location = cache.store("complete fetched content").await.unwrap().unwrap(); + let name = location.rsplit('/').next().unwrap(); + let id = name.strip_suffix(".txt").unwrap(); + assert!(Uuid::parse_str(id).is_ok()); + assert_eq!( + std::fs::read_to_string(cache.directory().join(name)).unwrap(), + "complete fetched content" + ); + assert_eq!( + location, + format!("http://127.0.0.1:4312/web-cache/{name}") + ); + + let response = cache + .router() + .oneshot( + Request::get(format!("/web-cache/{name}")) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + assert_eq!( + to_bytes(response.into_body(), usize::MAX).await.unwrap(), + "complete fetched content" + ); + } +} diff --git a/server/src/search/fetch.rs b/server/src/search/fetch.rs index 5d99d10..fbf7fd5 100644 --- a/server/src/search/fetch.rs +++ b/server/src/search/fetch.rs @@ -15,7 +15,7 @@ use reqwest::{ use tokio::{net::lookup_host, time::timeout}; use url::{Host, Url}; -use crate::store::Store; +use crate::{search::WebCache, store::Store}; const MAX_RESPONSE_SIZE: usize = 5 * 1024 * 1024; const MAX_REDIRECTS: usize = 5; diff --git a/server/src/search/mod.rs b/server/src/search/mod.rs index 695104c..97e1876 100644 --- a/server/src/search/mod.rs +++ b/server/src/search/mod.rs @@ -1,10 +1,12 @@ //! Exposes provider-independent search capabilities. +mod cache; mod catalog; mod engine; mod federation; mod fetch; mod search_provider; +pub use cache::WebCache; pub use engine::{HtmlEngine, JsonEngine, SearchEngine, SearchHit}; pub use federation::{SearchError, WebSearch}; pub use fetch::{FetchError, FetchedPage, WebFetch}; diff --git a/server/src/store/migrations.rs b/server/src/store/migrations.rs new file mode 100644 index 0000000..01b47e9 --- /dev/null +++ b/server/src/store/migrations.rs @@ -0,0 +1,479 @@ +//! Runs embedded SQLite migrations with enough progress data to diagnose startup failures. +use std::{ + borrow::Cow, + collections::HashSet, + future::Future, + path::{Path, PathBuf}, + time::{Duration, Instant}, +}; + +use sqlx::{ + migrate::{Migration, Migrator}, + Row, SqlitePool, +}; + +use crate::{Error, Result}; + +const MIGRATION_HEARTBEAT_INTERVAL: Duration = Duration::from_secs(2); +const MIGRATION_STAGE_TIMEOUT: Duration = Duration::from_secs(30); + +static ALL_MIGRATIONS: Migrator = sqlx::migrate!("./migrations"); + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum MigrationLineEndings { + Lf, + Crlf, +} + +impl MigrationLineEndings { + fn label(self) -> &'static str { + match self { + Self::Lf => "lf", + Self::Crlf => "crlf", + } + } +} + +#[derive(Debug)] +struct AppliedMigration { + version: i64, + description: String, + success: bool, + checksum: Vec, + execution_time_ns: i64, +} + +fn migrator_with_line_endings(line_endings: MigrationLineEndings) -> Migrator { + let migrations = ALL_MIGRATIONS + .iter() + .map(|migration| { + let lf_sql = migration.sql.replace("\r\n", "\n"); + let sql = match line_endings { + MigrationLineEndings::Lf => lf_sql, + MigrationLineEndings::Crlf => lf_sql.replace('\n', "\r\n"), + }; + Migration::new( + migration.version, + migration.description.clone(), + migration.migration_type, + Cow::Owned(sql), + migration.no_tx, + ) + }) + .collect::>(); + Migrator { + migrations: Cow::Owned(migrations), + ..Migrator::DEFAULT + } +} + +fn history_matches(migrator: &Migrator, applied: &[AppliedMigration]) -> bool { + applied.iter().all(|applied| { + applied.success + && migrator.iter().any(|known| { + known.version == applied.version && known.checksum.as_ref() == applied.checksum + }) + }) +} + +fn select_migrator(applied: &[AppliedMigration]) -> (Migrator, MigrationLineEndings) { + let lf = migrator_with_line_endings(MigrationLineEndings::Lf); + if history_matches(&lf, applied) { + return (lf, MigrationLineEndings::Lf); + } + + let crlf = migrator_with_line_endings(MigrationLineEndings::Crlf); + if history_matches(&crlf, applied) { + return (crlf, MigrationLineEndings::Crlf); + } + + // Preserve SQLx's exact validation error for dirty, unknown, or genuinely modified history. + (lf, MigrationLineEndings::Lf) +} + +pub(super) async fn run(pool: &SqlitePool, database_path: &Path) -> Result<()> { + tracing::info!("running database migrations with diagnostics"); + log_database_files(database_path); + log_sqlite_state(pool).await?; + + let applied = load_applied_migrations(pool).await?; + let (migrator, line_endings) = select_migrator(&applied); + tracing::info!( + line_endings = line_endings.label(), + "selected database migration line endings" + ); + let known_versions = migrator + .iter() + .map(|migration| migration.version) + .collect::>(); + let applied_versions = applied + .iter() + .map(|migration| migration.version) + .collect::>(); + let known = migrator + .iter() + .map(|migration| format!("{:04} {}", migration.version, migration.description)) + .collect::>(); + let pending = migrator + .iter() + .filter(|migration| !applied_versions.contains(&migration.version)) + .map(|migration| format!("{:04} {}", migration.version, migration.description)) + .collect::>(); + + tracing::info!(known = ?known, "embedded database migrations"); + if applied.is_empty() { + tracing::info!("no applied database migrations were found"); + } else { + for migration in &applied { + tracing::info!( + version = migration.version, + description = %migration.description, + success = migration.success, + execution_time_ns = migration.execution_time_ns, + checksum_bytes = migration.checksum.len(), + "applied database migration" + ); + } + } + tracing::info!(pending = ?pending, pending_count = pending.len(), "pending database migrations"); + + let invalid_history = applied.iter().any(|applied| { + let Some(known) = migrator + .iter() + .find(|migration| migration.version == applied.version) + else { + return true; + }; + !applied.success || known.checksum.as_ref() != applied.checksum + }) || applied + .iter() + .any(|migration| !known_versions.contains(&migration.version)); + + if invalid_history { + tracing::error!( + "database migration history is dirty, unknown, or has a checksum mismatch; running SQLx validation for the exact error" + ); + return run_stage("migration history validation", migrator.run(pool)).await; + } + + for (index, migration) in migrator.iter().enumerate() { + if applied_versions.contains(&migration.version) { + continue; + } + + let stage = format!("{:04} {}", migration.version, migration.description); + let prefix = Migrator { + migrations: Cow::Owned(migrator.iter().take(index + 1).cloned().collect()), + ignore_missing: true, + ..Migrator::DEFAULT + }; + run_stage(&stage, prefix.run(pool)).await?; + log_database_files(database_path); + log_sqlite_state(pool).await?; + } + + run_stage("final migration history validation", migrator.run(pool)).await?; + tracing::info!("database migrations completed"); + Ok(()) +} + +async fn run_stage(stage: &str, future: F) -> Result<()> +where + F: Future>, +{ + run_stage_with_limits( + stage, + MIGRATION_HEARTBEAT_INTERVAL, + MIGRATION_STAGE_TIMEOUT, + future, + ) + .await +} + +async fn run_stage_with_limits( + stage: &str, + heartbeat_interval: Duration, + stage_timeout: Duration, + future: F, +) -> Result<()> +where + F: Future>, +{ + let started = Instant::now(); + let mut heartbeat = tokio::time::interval(heartbeat_interval); + heartbeat.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + heartbeat.tick().await; + let deadline = tokio::time::sleep(stage_timeout); + tokio::pin!(deadline); + tokio::pin!(future); + + tracing::info!( + stage, + timeout_seconds = stage_timeout.as_secs_f64(), + "database migration stage started" + ); + loop { + tokio::select! { + result = &mut future => { + let elapsed_ms = started.elapsed().as_millis(); + return match result { + Ok(()) => { + tracing::info!(stage, elapsed_ms, "database migration stage completed"); + Ok(()) + } + Err(error) => { + tracing::error!(stage, elapsed_ms, %error, "database migration stage failed"); + Err(error.into()) + } + }; + } + _ = heartbeat.tick() => { + tracing::warn!( + stage, + elapsed_ms = started.elapsed().as_millis(), + "database migration stage is still running; the database may be locked by another process" + ); + } + _ = &mut deadline => { + let elapsed_seconds = started.elapsed().as_secs(); + tracing::error!(stage, elapsed_seconds, "database migration stage timed out"); + return Err(Error::MigrationTimeout { + stage: stage.to_owned(), + timeout_seconds: stage_timeout.as_secs(), + }); + } + } + } +} + +async fn load_applied_migrations(pool: &SqlitePool) -> Result> { + tracing::info!("reading database migration history"); + let table_exists: i64 = sqlx::query_scalar( + "SELECT EXISTS( + SELECT 1 FROM sqlite_master + WHERE type = 'table' AND name = '_sqlx_migrations' + )", + ) + .fetch_one(pool) + .await?; + if table_exists == 0 { + tracing::info!("database migration history table does not exist"); + return Ok(Vec::new()); + } + + let rows = sqlx::query( + "SELECT version, description, success, checksum, execution_time + FROM _sqlx_migrations ORDER BY version", + ) + .fetch_all(pool) + .await?; + tracing::info!(row_count = rows.len(), "database migration history loaded"); + rows.into_iter() + .map(|row| { + Ok(AppliedMigration { + version: row.try_get("version")?, + description: row.try_get("description")?, + success: row.try_get("success")?, + checksum: row.try_get("checksum")?, + execution_time_ns: row.try_get("execution_time")?, + }) + }) + .collect::, sqlx::Error>>() + .map_err(Into::into) +} + +async fn log_sqlite_state(pool: &SqlitePool) -> Result<()> { + tracing::info!( + pool_size = pool.size(), + pool_idle = pool.num_idle(), + "reading SQLite runtime state" + ); + let sqlite_version: String = sqlx::query_scalar("SELECT sqlite_version()") + .fetch_one(pool) + .await?; + let journal_mode: String = sqlx::query_scalar("PRAGMA journal_mode") + .fetch_one(pool) + .await?; + let locking_mode: String = sqlx::query_scalar("PRAGMA locking_mode") + .fetch_one(pool) + .await?; + let foreign_keys: i64 = sqlx::query_scalar("PRAGMA foreign_keys") + .fetch_one(pool) + .await?; + let busy_timeout_ms: i64 = sqlx::query_scalar("PRAGMA busy_timeout") + .fetch_one(pool) + .await?; + let schema_version: i64 = sqlx::query_scalar("PRAGMA schema_version") + .fetch_one(pool) + .await?; + let user_version: i64 = sqlx::query_scalar("PRAGMA user_version") + .fetch_one(pool) + .await?; + let page_size: i64 = sqlx::query_scalar("PRAGMA page_size") + .fetch_one(pool) + .await?; + let page_count: i64 = sqlx::query_scalar("PRAGMA page_count") + .fetch_one(pool) + .await?; + let freelist_count: i64 = sqlx::query_scalar("PRAGMA freelist_count") + .fetch_one(pool) + .await?; + + tracing::info!( + %sqlite_version, + %journal_mode, + %locking_mode, + foreign_keys, + busy_timeout_ms, + schema_version, + user_version, + page_size, + page_count, + freelist_count, + "SQLite runtime state" + ); + Ok(()) +} + +fn log_database_files(database_path: &Path) { + tracing::info!( + database_path = %database_path.display(), + process_id = std::process::id(), + "SQLite database files" + ); + log_file("database", database_path); + log_file("wal", &sidecar_path(database_path, "-wal")); + log_file("shm", &sidecar_path(database_path, "-shm")); + log_file("journal", &sidecar_path(database_path, "-journal")); +} + +fn log_file(kind: &str, path: &Path) { + match std::fs::metadata(path) { + Ok(metadata) => tracing::info!( + kind, + path = %path.display(), + exists = true, + size_bytes = metadata.len(), + readonly = metadata.permissions().readonly(), + "SQLite file state" + ), + Err(error) if error.kind() == std::io::ErrorKind::NotFound => tracing::info!( + kind, + path = %path.display(), + exists = false, + "SQLite file state" + ), + Err(error) => { + tracing::warn!(kind, path = %path.display(), %error, "failed to read SQLite file state") + } + } +} + +fn sidecar_path(database_path: &Path, suffix: &str) -> PathBuf { + let mut path = database_path.as_os_str().to_os_string(); + path.push(suffix); + PathBuf::from(path) +} + +#[cfg(test)] +mod tests { + use sqlx::sqlite::SqlitePoolOptions; + + use super::*; + + #[test] + fn normalized_checksums_match_the_published_lf_and_windows_crlf_migrations() { + let lf = migrator_with_line_endings(MigrationLineEndings::Lf); + let crlf = migrator_with_line_endings(MigrationLineEndings::Crlf); + let lf_initial = lf.iter().find(|migration| migration.version == 1).unwrap(); + let crlf_initial = crlf + .iter() + .find(|migration| migration.version == 1) + .unwrap(); + + assert_eq!( + hex::encode(lf_initial.checksum.as_ref()), + "ddf1bc573e460bfd93ea50b60d003e8fc6bb9a1b32de71139cb0fc0d898e88c401c08c8a8b57abbe68f55927e7f004d9" + ); + assert_eq!( + hex::encode(crlf_initial.checksum.as_ref()), + "7c5995693dbd5f9d50880fc874784cb67c499762abcee4a562b54c9afd8239bae125074696b78f90aca0fc136b802a2b" + ); + } + + #[tokio::test] + async fn lf_and_crlf_migration_histories_upgrade_without_rewriting_checksums() { + for line_endings in [MigrationLineEndings::Lf, MigrationLineEndings::Crlf] { + let pool = SqlitePoolOptions::new() + .max_connections(1) + .connect("sqlite::memory:") + .await + .unwrap(); + let historical = migrator_with_line_endings(line_endings); + let first_four = Migrator { + migrations: Cow::Owned(historical.iter().take(4).cloned().collect()), + ..Migrator::DEFAULT + }; + first_four.run(&pool).await.unwrap(); + let checksum_before: Vec = + sqlx::query_scalar("SELECT checksum FROM _sqlx_migrations WHERE version = 1") + .fetch_one(&pool) + .await + .unwrap(); + + run(&pool, Path::new("line-ending-compatibility.db")) + .await + .unwrap(); + + let checksum_after: Vec = + sqlx::query_scalar("SELECT checksum FROM _sqlx_migrations WHERE version = 1") + .fetch_one(&pool) + .await + .unwrap(); + let versions: Vec = + sqlx::query_scalar("SELECT version FROM _sqlx_migrations ORDER BY version") + .fetch_all(&pool) + .await + .unwrap(); + let checkpoint_table_exists: i64 = sqlx::query_scalar( + "SELECT EXISTS( + SELECT 1 FROM sqlite_master + WHERE type = 'table' AND name = 'conversation_checkpoints' + )", + ) + .fetch_one(&pool) + .await + .unwrap(); + + assert_eq!(checksum_after, checksum_before); + assert_eq!(versions, vec![1, 2, 3, 4, 5, 6]); + assert_eq!(checkpoint_table_exists, 1); + } + } + + #[test] + fn sqlite_sidecar_paths_preserve_the_database_path() { + let database = Path::new(r"C:\Users\Test User\cursor-byok.db"); + assert_eq!( + sidecar_path(database, "-wal"), + PathBuf::from(r"C:\Users\Test User\cursor-byok.db-wal") + ); + } + + #[tokio::test] + async fn stalled_migration_stage_returns_a_timeout_error() { + let result = run_stage_with_limits( + "0006 rename revisions to checkpoints", + Duration::from_millis(2), + Duration::from_millis(10), + std::future::pending(), + ) + .await; + + assert!(matches!( + result, + Err(Error::MigrationTimeout { stage, .. }) + if stage == "0006 rename revisions to checkpoints" + )); + } +} diff --git a/server/src/store/mod.rs b/server/src/store/mod.rs index 7e6e5d2..b9a0bf5 100644 --- a/server/src/store/mod.rs +++ b/server/src/store/mod.rs @@ -7,6 +7,7 @@ mod input_anchors; mod legacy_config; mod llm_calls; mod messages; +mod migrations; mod models; mod overview; mod runs; diff --git a/server/src/store/sqlite.rs b/server/src/store/sqlite.rs index 691cd71..1403cf8 100644 --- a/server/src/store/sqlite.rs +++ b/server/src/store/sqlite.rs @@ -8,7 +8,7 @@ use sqlx::{ use crate::Result; -use super::writer::WriteCoordinator; +use super::{migrations, writer::WriteCoordinator}; #[derive(Clone)] pub struct Store { @@ -24,13 +24,12 @@ impl Store { .journal_mode(SqliteJournalMode::Wal) .synchronous(SqliteSynchronous::Full) .busy_timeout(Duration::from_secs(5)); + let database_path = options.get_filename().to_owned(); let pool = SqlitePoolOptions::new() .max_connections(8) .connect_with(options) .await?; - tracing::info!("running database migrations"); - sqlx::migrate!("./migrations").run(&pool).await?; - tracing::info!("database migrations completed"); + migrations::run(&pool, &database_path).await?; Ok(Self { pool, writes: WriteCoordinator::default(), diff --git a/server/tests/checkpoint_recovery.rs b/server/tests/checkpoint_recovery.rs index 4ef5f45..39f7115 100644 --- a/server/tests/checkpoint_recovery.rs +++ b/server/tests/checkpoint_recovery.rs @@ -19,7 +19,7 @@ use cursor_server::{ use prost::Message; #[tokio::test] -async fn applied_revision_schema_upgrades_to_checkpoints_without_losing_rows() { +async fn v0_1_5_beta_1_schema_upgrades_to_checkpoints_without_losing_rows() { use std::borrow::Cow; use sqlx::{ @@ -42,11 +42,12 @@ async fn applied_revision_schema_upgrades_to_checkpoints_without_losing_rows() { ) .await .unwrap(); - let previous = Migrator { - migrations: Cow::Owned(ALL_MIGRATIONS.iter().take(5).cloned().collect()), + // v0.1.5-beta.1 shipped migrations 0001 through 0004. + let v0_1_5_beta_1 = Migrator { + migrations: Cow::Owned(ALL_MIGRATIONS.iter().take(4).cloned().collect()), ..Migrator::DEFAULT }; - previous.run(&pool).await.unwrap(); + v0_1_5_beta_1.run(&pool).await.unwrap(); sqlx::query( "INSERT INTO conversations(conversation_id, updated_at_ms)