Repository: tursodatabase/libsql (c)

Issue:
Sync overwrites local databases
If I have a local database file `local.db` that has some data and I configure a sync URL to an unrelated remote database, the sync operation will overwrite the local database file. We perhaps should make sync work so that we'll only sync if there is an existing WAL index file or there's no local database file to avoid people overwriting their data accidentally.

Pull request:
libsql: prevent database overwrite misuse
Closes #1145

Patch:
diff --git a/libsql-ffi/build.rs b/libsql-ffi/build.rs
index fd631aabfe..2b0aa80897 100644
--- a/libsql-ffi/build.rs
+++ b/libsql-ffi/build.rs
@@ -339,6 +339,11 @@ fn build_multiple_ciphers(out_path: &Path) {
     cmake_opts.push("-DSQLITE_USE_URI=ON");
     cmake_opts.push("-DCMAKE_POSITION_INDEPENDENT_CODE=ON");
 
+    if cargo_build_target.contains("musl") {
+        cmake_opts.push("-DCMAKE_C_FLAGS=\"-U_FORTIFY_SOURCE\"");
+        cmake_opts.push("-DCMAKE_CXX_FLAGS=\"-U_FORTIFY_SOURCE\"");
+    }
+
     let mut cmake = Command::new("cmake");
     cmake.current_dir("bundled/SQLite3MultipleCiphers/build");
     cmake.args(cmake_opts.clone());
diff --git a/libsql-replication/src/meta.rs b/libsql-replication/src/meta.rs
index 782ce76b6d..2a617a9a96 100644
--- a/libsql-replication/src/meta.rs
+++ b/libsql-replication/src/meta.rs
@@ -27,6 +27,10 @@ pub enum Error {
     FailedToCommit(std::io::Error),
     #[error("Invalid replication path")]
     InvalidReplicationPath,
+    #[error(
+        "Can not sync a database without a wal_index, please delete the database and attempt again"
+    )]
+    RequiresCleanDatabase,
 }
 
 #[repr(C)]
@@ -86,6 +90,13 @@ impl WalIndexMeta {
 
         let path = db_path.with_file_name(wal_index_file);
 
+        // If there is no database or there exists a database AND a wal index file
+        // then allow the embedded replica to be created. If Neither of those conditions are met
+        // for example a database without a index file then we throw this error.
+        if !(!db_path.exists() || path.exists()) {
+            return Err(Error::RequiresCleanDatabase);
+        }
+
         let mut file = tokio::fs::OpenOptions::new()
             .create(true)
             .read(true)
diff --git a/libsql-server/src/http/admin/mod.rs b/libsql-server/src/http/admin/mod.rs
index a36e9806f8..635537b7b3 100644
--- a/libsql-server/src/http/admin/mod.rs
+++ b/libsql-server/src/http/admin/mod.rs
@@ -306,6 +306,12 @@ async fn handle_create_namespace<C: Connector>(
         config.jwt_key = Some(jwt_key);
     }
 
+    if req.shared_schema_name.is_some() && req.dump_url.is_some() {
+        return Err(Error::SharedSchemaCreationError(
+            "shared schema database cannot be created from a dump".to_string(),
+        ));
+    }
+
     if let Some(ns) = req.shared_schema_name {
         if req.shared_schema {
             return Err(Error::SharedSchemaCreationError(
diff --git a/libsql-server/src/namespace/meta_store.rs b/libsql-server/src/namespace/meta_store.rs
index 6f8df63ce2..c2d14650f7 100644
--- a/libsql-server/src/namespace/meta_store.rs
+++ b/libsql-server/src/namespace/meta_store.rs
@@ -5,6 +5,7 @@ use std::{collections::HashMap, fs::read_dir};
 
 use bottomless::bottomless_wal::BottomlessWalWrapper;
 use bottomless::replicator::CompressionKind;
+use bottomless::SavepointTracker;
 use futures_core::Future;
 use libsql_replication::rpc::metadata;
 use libsql_sys::wal::{
@@ -486,6 +487,14 @@ impl MetaStore {
         .unwrap()?;
         Ok(details)
     }
+
+    pub fn backup_savepoint(&self) -> Option<SavepointTracker> {
+        let lock = self.inner.lock();
+        if let Some(wal) = lock.wal_manager.wrapper() {
+            return wal.backup_savepoint();
+        }
+        None
+    }
 }
 
 impl MetaStoreHandle {
diff --git a/libsql-server/src/namespace/mod.rs b/libsql-server/src/namespace/mod.rs
index dc333e12aa..a9f09d7f42 100644
--- a/libsql-server/src/namespace/mod.rs
+++ b/libsql-server/src/namespace/mod.rs
@@ -477,7 +477,8 @@ impl Namespace {
                             | Error::Io(_)
                             | Error::InvalidLogId
                             | Error::FailedToCommit(_)
-                            | Error::InvalidReplicationPath => {
+                            | Error::InvalidReplicationPath
+                            | Error::RequiresCleanDatabase => {
                                 // We retry from last frame index?
                                 tracing::warn!("non-fatal replication error, retrying from last commit index: {err}");
                             },
diff --git a/libsql-server/src/schema/db.rs b/libsql-server/src/schema/db.rs
index 8cc6525c7b..212c7239d3 100644
--- a/libsql-server/src/schema/db.rs
+++ b/libsql-server/src/schema/db.rs
@@ -274,7 +274,6 @@ pub(super) fn get_next_pending_migration_job(
                     job_id,
                     status,
                     migration,
-                    backup_sync: None,
                     progress: Default::default(),
                 })
             },
diff --git a/libsql-server/src/schema/scheduler.rs b/libsql-server/src/schema/scheduler.rs
index f2b1230181..704de60e82 100644
--- a/libsql-server/src/schema/scheduler.rs
+++ b/libsql-server/src/schema/scheduler.rs
@@ -174,40 +174,32 @@ impl Scheduler {
                         let old_status = *job.status();
                         let migration_db = self.migration_db.clone();
                         let store = self.namespace_store.clone();
-                        let schema = job.schema();
                         async move {
-                            let mut status = with_conn_async(migration_db,
-                                move |conn| job_step_dry_run_success(conn, job_id)
-                            )
-                                .await
-                                .unwrap()
-                                .unwrap_or(old_status);
+                            let mut status = with_conn_async(migration_db, move |conn| {
+                                job_step_dry_run_success(conn, job_id)
+                            })
+                            .await
+                            .unwrap()
+                            .unwrap_or(old_status);
 
                             if status.is_dry_run_success() {
-                                let savepoint = store
-                                    .with(schema, |ns| {
-                                        ns.db
-                                            .as_schema()
-                                            .expect("expected database to be a schema database")
-                                            .backup_savepoint()
-                                    })
-                                .await
-                                    .unwrap();
-
-                                if let Some(mut savepoint) = savepoint {
+                                if let Some(mut savepoint) = store.meta_store().backup_savepoint() {
                                     if let Err(e) = savepoint.confirmed().await {
-                                        tracing::error!("failed to backup metastore after state change on job {job_id}: {e}");
+                                        tracing::error!("failed to backup meta store: {e}");
+                                        // do not step the job, and schedule for retry.
+                                        // TODO: backoff?
+                                        // TODO: this is fine if we don't manage to get a backup here,
+                                        // then we'll restart in the previous state in case of restore,
+                                        // however, in case of restart we may not have a backup.
                                         status = MigrationJobStatus::WaitingRun;
                                     }
                                 }
-
                                 // todo!("notify dry run success")
-                                // nothing more to do in this call, return early and let next call enqueue
-                                // step the job
                             }
 
                             WorkResult::Job { status }
-                        }});
+                        }
+                    });
 
                     *job.status_mut() = MigrationJobStatus::WaitingTransition;
                     self.has_work = false;
@@ -216,19 +208,33 @@ impl Scheduler {
             }
             MigrationJobStatus::DryRunSuccess => {
                 let job_id = job.job_id();
-                with_conn_async(self.migration_db.clone(), move |conn| {
-                    update_job_status(conn, job_id, MigrationJobStatus::WaitingRun)
-                })
-                .await
-                .unwrap();
-                let new_status = match job.wait_for_backup().await {
-                    Ok(_) => MigrationJobStatus::WaitingRun,
-                    Err(_backup_err) => {
-                        // todo!("backup for schema db failed because: {}", _backup_err);
-                        MigrationJobStatus::DryRunFailure
+                let store = self.namespace_store.clone();
+                let migration_db = self.migration_db.clone();
+                self.workers.spawn(async move {
+                    with_conn_async(migration_db, move |conn| {
+                        update_job_status(conn, job_id, MigrationJobStatus::WaitingRun)
+                    })
+                    .await
+                    .unwrap();
+
+                    let mut status = MigrationJobStatus::WaitingRun;
+                    if let Some(mut savepoint) = store.meta_store().backup_savepoint() {
+                        if let Err(e) = savepoint.confirmed().await {
+                            tracing::error!("failed to backup meta store: {e}");
+                            // do not step the job, and schedule for retry.
+                            // TODO: backoff?
+                            // TODO: this is fine if we don't manage to get a backup here,
+                            // then we'll restart in the previous state in case of restore,
+                            // however, in case of restart we may not have a backup.
+                            status = MigrationJobStatus::DryRunSuccess;
+                        }
                     }
-                };
-                *job.status_mut() = new_status;
+
+                    WorkResult::Job { status }
+                });
+
+                *job.status_mut() = MigrationJobStatus::WaitingTransition;
+                self.has_work = false;
             }
             MigrationJobStatus::DryRunFailure => todo!(),
             MigrationJobStatus::WaitingRun => {
@@ -260,6 +266,8 @@ impl Scheduler {
                         let job_id = job.job_id();
                         let migration = job.migration();
                         let migration_db = self.migration_db.clone();
+                        let store = self.namespace_store.clone();
+                        let schema = job.schema();
                         async move {
                             let _perm = permit;
                             tokio::task::spawn_blocking(move || {
@@ -286,8 +294,29 @@ impl Scheduler {
                                         txn.commit().unwrap();
                                     }
                                 });
+                            }).await.unwrap();
+
+
+                            // backup the schema
+                            let savepoint = store
+                                .with(schema, |ns| {
+                                    ns.db
+                                        .as_schema()
+                                        .expect("expected database to be a schema database")
+                                        .backup_savepoint()
+                                })
+                            .await
+                                .unwrap();
+
+                            if let Some(mut savepoint) = savepoint {
+                                if let Err(e) = savepoint.confirmed().await {
+                                    tracing::error!("failed to backup metastore after state change on job {job_id}: {e}");
+                                    // early return, we couldn't backup the schema
+                                    return WorkResult::Job { status: MigrationJobStatus::WaitingRun };
+                                }
+                            }
 
-                                {
+                            tokio::task::spawn_blocking(move || {
                                     let mut conn = migration_db.lock();
                                     update_job_status(
                                         &mut conn,
@@ -295,14 +324,24 @@ impl Scheduler {
                                         MigrationJobStatus::RunSuccess,
                                     )
                                     .unwrap();
-                                }
-
-                                WorkResult::Job {
-                                    status: MigrationJobStatus::RunSuccess,
-                                }
                             })
                             .await
-                            .unwrap()
+                            .unwrap();
+
+                            let mut status = MigrationJobStatus::RunSuccess;
+                            if let Some(mut savepoint) = store.meta_store().backup_savepoint() {
+                                if let Err(e) = savepoint.confirmed().await {
+                                    tracing::error!("failed to backup meta store: {e}");
+                                    // do not step the job, and schedule for retry.
+                                    // TODO: backoff?
+                                    // TODO: this is fine if we don't manage to get a backup here,
+                                    // then we'll restart in the previous state in case of restore,
+                                    // however, in case of restart we may not have a backup.
+                                    status = MigrationJobStatus::WaitingRun;
+                                }
+                            }
+
+                            WorkResult::Job { status }
                         }
                     });
                     // do not enqueue anything until the schema migration is complete
diff --git a/libsql-server/src/schema/status.rs b/libsql-server/src/schema/status.rs
index f7cbcb9992..24876fb225 100644
--- a/libsql-server/src/schema/status.rs
+++ b/libsql-server/src/schema/status.rs
@@ -1,6 +1,5 @@
 use std::sync::Arc;
 
-use bottomless::SavepointTracker;
 use serde::{Deserialize, Serialize};
 
 use crate::{connection::program::Program, namespace::NamespaceName};
@@ -37,7 +36,6 @@ pub struct MigrationJob {
     pub(super) status: MigrationJobStatus,
     pub(super) job_id: i64,
     pub(super) migration: Arc<Program>,
-    pub(super) backup_sync: Option<Arc<SavepointTracker>>,
     pub(super) progress: [usize; MigrationTaskStatus::num_variants()],
 }
 
@@ -96,14 +94,6 @@ impl MigrationJob {
             MigrationJobStatus::WaitingTransition => 0,
         }
     }
-
-    pub(crate) async fn wait_for_backup(&mut self) -> crate::Result<()> {
-        if let Some(mut savepoint) = self.backup_sync.take() {
-            let savepoint = Arc::get_mut(&mut savepoint).unwrap();
-            savepoint.confirmed().await?;
-        }
-        Ok(())
-    }
 }
 
 /// Represents the status of a migration task
diff --git a/libsql/src/database.rs b/libsql/src/database.rs
index 48cf950e75..45234f0b88 100644
--- a/libsql/src/database.rs
+++ b/libsql/src/database.rs
@@ -592,3 +592,9 @@ fn connector() -> hyper_rustls::HttpsConnector<hyper::client::HttpConnector> {
         .enable_http1()
         .wrap_connector(http)
 }
+
+impl std::fmt::Debug for Database {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("Database").finish()
+    }
+}
diff --git a/libsql/src/database/builder.rs b/libsql/src/database/builder.rs
index cb8cf15e42..5eb7ab8594 100644
--- a/libsql/src/database/builder.rs
+++ b/libsql/src/database/builder.rs
@@ -17,6 +17,12 @@ use super::DbType;
 ///     includes the ability to delegate writes to a remote primary.
 /// - `new_remote`/`Remote` creates a database that does not create anything locally but will
 ///     instead run all queries on the remote database. This is essentially the pure HTTP api.
+///
+/// # Note
+///
+/// Embedded replica's require a clean database (no database file) or a previously synced database or else it will
+/// throw an error to prevent any misuse. To work around this error a user can delete the database
+/// and let it resync and create the wal_index metadata file.
 pub struct Builder<T = ()> {
     inner: T,
 }


Test patch:
diff --git a/libsql-server/src/schema/snapshots/libsql_server__schema__db__test__pending_job-3.snap b/libsql-server/src/schema/snapshots/libsql_server__schema__db__test__pending_job-3.snap
index ecff9063fe..e300b7c8d7 100644
--- a/libsql-server/src/schema/snapshots/libsql_server__schema__db__test__pending_job-3.snap
+++ b/libsql-server/src/schema/snapshots/libsql_server__schema__db__test__pending_job-3.snap
@@ -26,7 +26,6 @@ MigrationJob {
             },
         ],
     },
-    backup_sync: None,
     progress: [
         2,
         0,
diff --git a/libsql-server/src/schema/snapshots/libsql_server__schema__db__test__pending_job.snap b/libsql-server/src/schema/snapshots/libsql_server__schema__db__test__pending_job.snap
index 8ddd58990f..a4c9ab181a 100644
--- a/libsql-server/src/schema/snapshots/libsql_server__schema__db__test__pending_job.snap
+++ b/libsql-server/src/schema/snapshots/libsql_server__schema__db__test__pending_job.snap
@@ -26,7 +26,6 @@ MigrationJob {
             },
         ],
     },
-    backup_sync: None,
     progress: [
         3,
         0,
diff --git a/libsql-server/tests/embedded_replica/mod.rs b/libsql-server/tests/embedded_replica/mod.rs
index 17dfd0b315..5d07aae896 100644
--- a/libsql-server/tests/embedded_replica/mod.rs
+++ b/libsql-server/tests/embedded_replica/mod.rs
@@ -895,3 +895,77 @@ fn sync_interval() {
 
     sim.run().unwrap();
 }
+
+#[test]
+fn errors_on_bad_replica() {
+    let mut sim = Builder::new()
+        .simulation_duration(Duration::from_secs(u64::MAX))
+        .build();
+
+    let tmp_embedded = tempdir().unwrap();
+    let tmp_host = tempdir().unwrap();
+    let tmp_embedded_path = tmp_embedded.path().to_owned();
+    let tmp_host_path = tmp_host.path().to_owned();
+
+    make_primary(&mut sim, tmp_host_path.clone());
+
+    sim.client("client", async move {
+        let client = Client::new();
+        client
+            .post("http://primary:9090/v1/namespaces/foo/create", json!({}))
+            .await?;
+
+        let path = tmp_embedded_path.join("embedded");
+        let db = libsql::Builder::new_remote_replica(
+            path.to_str().unwrap(),
+            "http://foo.primary:8080".to_string(),
+            "".to_string(),
+        )
+        .connector(TurmoilConnector)
+        .build()
+        .await?;
+
+        let conn = db.connect()?;
+
+        conn.execute("create table test (x)", ()).await?;
+
+        conn.execute("insert into test values (12)", ())
+            .await
+            .unwrap();
+
+        db.sync().await.unwrap();
+
+        drop(conn);
+        drop(db);
+
+        let wal_index_file = format!("{}-client_wal_index", path.to_str().unwrap());
+
+        std::fs::remove_file(wal_index_file).unwrap();
+
+        libsql::Builder::new_remote_replica(
+            path.to_str().unwrap(),
+            "http://foo.primary:8080".to_string(),
+            "".to_string(),
+        )
+        .connector(TurmoilConnector)
+        .build()
+        .await
+        .unwrap_err();
+
+        std::fs::remove_file(&path).unwrap();
+
+        libsql::Builder::new_remote_replica(
+            path.to_str().unwrap(),
+            "http://foo.primary:8080".to_string(),
+            "".to_string(),
+        )
+        .connector(TurmoilConnector)
+        .build()
+        .await
+        .unwrap();
+
+        Ok(())
+    });
+
+    sim.run().unwrap();
+}