Merge pull request #1718 from tursodatabase/schmema-db-replica-regres…

…sion Fix schema db bug with replicas
tursodatabase · Sep 2, 2024 · 73a2913 · 73a2913
2 parents d8a4e40 + f0c92ca
commit 73a2913
Show file tree

Hide file tree

Showing 10 changed files with 336 additions and 68 deletions.
diff --git a/libsql-server/src/connection/connection_core.rs b/libsql-server/src/connection/connection_core.rs
@@ -659,9 +659,15 @@ mod test {
         let ctx = RequestContext::new(
             Authenticated::FullAccess,
             NamespaceName::default(),
-            MetaStore::new(Default::default(), tmp.path(), maker().unwrap(), manager)
-                .await
-                .unwrap(),
+            MetaStore::new(
+                Default::default(),
+                tmp.path(),
+                maker().unwrap(),
+                manager,
+                crate::database::DatabaseKind::Primary,
+            )
+            .await
+            .unwrap(),
         );
         conn.execute_program(
             Program::seq(&["CREATE TABLE test (x)"]),

diff --git a/libsql-server/src/lib.rs b/libsql-server/src/lib.rs
@@ -570,6 +570,7 @@ where
             &self.path,
             meta_conn,
             meta_store_wal_manager,
+            db_kind,
         )
         .await?;
 

diff --git a/libsql-server/src/namespace/configurator/replica.rs b/libsql-server/src/namespace/configurator/replica.rs
@@ -49,6 +49,7 @@ impl ReplicaConfigurator {
 }
 
 impl ConfigureNamespace for ReplicaConfigurator {
+    #[tracing::instrument(skip_all, fields(name))]
     fn setup<'a>(
         &'a self,
         meta_store_handle: MetaStoreHandle,

diff --git a/libsql-server/src/namespace/meta_store.rs b/libsql-server/src/namespace/meta_store.rs
@@ -23,6 +23,7 @@ use tokio::sync::{
 
 use crate::config::BottomlessConfig;
 use crate::connection::config::DatabaseConfig;
+use crate::database::DatabaseKind;
 use crate::schema::{MigrationDetails, MigrationSummary};
 use crate::{
     config::MetaStoreConfig, connection::legacy::open_conn_active_checkpoint, error::Error, Result,
@@ -74,6 +75,7 @@ struct MetaStoreInner {
     configs: tokio::sync::Mutex<HashMap<NamespaceName, Sender<InnerConfig>>>,
     conn: tokio::sync::Mutex<MetaStoreConnection>,
     wal_manager: MetaStoreWalManager,
+    db_kind: DatabaseKind,
 }
 
 fn setup_connection(conn: &rusqlite::Connection) -> Result<()> {
@@ -182,13 +184,15 @@ impl MetaStoreInner {
         conn: MetaStoreConnection,
         wal_manager: MetaStoreWalManager,
         config: MetaStoreConfig,
+        db_kind: DatabaseKind,
     ) -> Result<Self> {
         setup_connection(&conn)?;
 
         let mut this = MetaStoreInner {
             configs: Default::default(),
             conn: conn.into(),
             wal_manager,
+            db_kind,
         };
 
         if config.allow_recover_from_fs {
@@ -350,9 +354,11 @@ fn try_process(
     let mut conn = inner.conn.blocking_lock();
     if let Some(schema) = config.shared_schema_name.as_ref() {
         let tx = conn.transaction()?;
-        if let Some(ref schema) = config.shared_schema_name {
-            if crate::schema::db::has_pending_migration_jobs(&tx, schema)? {
-                return Err(crate::Error::PendingMigrationOnSchema(schema.clone()));
+        if inner.db_kind.is_primary() {
+            if let Some(ref schema) = config.shared_schema_name {
+                if crate::schema::db::has_pending_migration_jobs(&tx, schema)? {
+                    return Err(crate::Error::PendingMigrationOnSchema(schema.clone()));
+                }
             }
         }
         tx.execute(
@@ -394,6 +400,7 @@ impl MetaStore {
         base_path: &Path,
         conn: MetaStoreConnection,
         wal_manager: MetaStoreWalManager,
+        db_kind: DatabaseKind,
     ) -> Result<Self> {
         let (changes_tx, mut changes_rx) = mpsc::channel(256);
 
@@ -402,7 +409,7 @@ impl MetaStore {
         let maybe_inner = tokio::task::spawn_blocking({
             let base_path = base_path.to_owned();
             let config = config.clone();
-            move || MetaStoreInner::new(&base_path, conn, wal_manager, config.clone())
+            move || MetaStoreInner::new(&base_path, conn, wal_manager, config.clone(), db_kind)
         })
         .await
         .unwrap();
@@ -446,7 +453,7 @@ impl MetaStore {
 
                     let inner = tokio::task::spawn_blocking({
                         let base_path = base_path.to_owned();
-                        move || MetaStoreInner::new(&base_path, conn, wal, config)
+                        move || MetaStoreInner::new(&base_path, conn, wal, config, db_kind)
                     })
                     .await
                     .unwrap()?;

diff --git a/libsql-server/src/replication/replicator_client.rs b/libsql-server/src/replication/replicator_client.rs
@@ -121,6 +121,7 @@ impl ReplicatorClient for Client {
                     .await
                     .map_err(|e| Status::new(Code::Internal, e.to_string()))?;
             }
+
             self.meta_store_handle
                 .store(DatabaseConfig::from(config))
                 .await

diff --git a/libsql-server/src/rpc/streaming_exec.rs b/libsql-server/src/rpc/streaming_exec.rs
@@ -396,9 +396,15 @@ pub mod test {
         let ctx = RequestContext::new(
             Authenticated::Anonymous,
             NamespaceName::default(),
-            MetaStore::new(Default::default(), tmp.path(), maker().unwrap(), manager)
-                .await
-                .unwrap(),
+            MetaStore::new(
+                Default::default(),
+                tmp.path(),
+                maker().unwrap(),
+                manager,
+                crate::database::DatabaseKind::Primary,
+            )
+            .await
+            .unwrap(),
         );
         let stream = make_proxy_stream(conn, ctx, ReceiverStream::new(rcv));
         pin!(stream);
@@ -422,9 +428,15 @@ pub mod test {
         let ctx = RequestContext::new(
             Authenticated::FullAccess,
             NamespaceName::default(),
-            MetaStore::new(Default::default(), tmp.path(), maker().unwrap(), manager)
-                .await
-                .unwrap(),
+            MetaStore::new(
+                Default::default(),
+                tmp.path(),
+                maker().unwrap(),
+                manager,
+                crate::database::DatabaseKind::Primary,
+            )
+            .await
+            .unwrap(),
         );
         let stream = make_proxy_stream(conn, ctx, ReceiverStream::new(rcv));
 
@@ -444,9 +456,15 @@ pub mod test {
         let ctx = RequestContext::new(
             Authenticated::FullAccess,
             NamespaceName::default(),
-            MetaStore::new(Default::default(), tmp.path(), maker().unwrap(), manager)
-                .await
-                .unwrap(),
+            MetaStore::new(
+                Default::default(),
+                tmp.path(),
+                maker().unwrap(),
+                manager,
+                crate::database::DatabaseKind::Primary,
+            )
+            .await
+            .unwrap(),
         );
         let stream = make_proxy_stream(conn, ctx, ReceiverStream::new(rcv));
 
@@ -468,9 +486,15 @@ pub mod test {
         let ctx = RequestContext::new(
             Authenticated::FullAccess,
             NamespaceName::default(),
-            MetaStore::new(Default::default(), tmp.path(), maker().unwrap(), manager)
-                .await
-                .unwrap(),
+            MetaStore::new(
+                Default::default(),
+                tmp.path(),
+                maker().unwrap(),
+                manager,
+                crate::database::DatabaseKind::Primary,
+            )
+            .await
+            .unwrap(),
         );
         // limit the size of the response to force a split
         let stream = make_proxy_stream_inner(conn, ctx, ReceiverStream::new(rcv), 500);
@@ -525,9 +549,15 @@ pub mod test {
         let ctx = RequestContext::new(
             Authenticated::FullAccess,
             NamespaceName::default(),
-            MetaStore::new(Default::default(), tmp.path(), maker().unwrap(), manager)
-                .await
-                .unwrap(),
+            MetaStore::new(
+                Default::default(),
+                tmp.path(),
+                maker().unwrap(),
+                manager,
+                crate::database::DatabaseKind::Primary,
+            )
+            .await
+            .unwrap(),
         );
         let stream = make_proxy_stream(conn, ctx, ReceiverStream::new(rcv));
 
@@ -552,9 +582,15 @@ pub mod test {
         let ctx = RequestContext::new(
             Authenticated::FullAccess,
             NamespaceName::default(),
-            MetaStore::new(Default::default(), tmp.path(), maker().unwrap(), manager)
-                .await
-                .unwrap(),
+            MetaStore::new(
+                Default::default(),
+                tmp.path(),
+                maker().unwrap(),
+                manager,
+                crate::database::DatabaseKind::Primary,
+            )
+            .await
+            .unwrap(),
         );
         let stream = make_proxy_stream(conn, ctx, ReceiverStream::new(rcv));
 
@@ -579,9 +615,15 @@ pub mod test {
         let ctx = RequestContext::new(
             Authenticated::FullAccess,
             NamespaceName::default(),
-            MetaStore::new(Default::default(), tmp.path(), maker().unwrap(), manager)
-                .await
-                .unwrap(),
+            MetaStore::new(
+                Default::default(),
+                tmp.path(),
+                maker().unwrap(),
+                manager,
+                crate::database::DatabaseKind::Primary,
+            )
+            .await
+            .unwrap(),
         );
         let stream = make_proxy_stream(conn, ctx, ReceiverStream::new(rcv));
 
@@ -608,9 +650,15 @@ pub mod test {
         let ctx = RequestContext::new(
             Authenticated::FullAccess,
             NamespaceName::default(),
-            MetaStore::new(Default::default(), tmp.path(), maker().unwrap(), manager)
-                .await
-                .unwrap(),
+            MetaStore::new(
+                Default::default(),
+                tmp.path(),
+                maker().unwrap(),
+                manager,
+                crate::database::DatabaseKind::Primary,
+            )
+            .await
+            .unwrap(),
         );
         let stream = make_proxy_stream(conn, ctx, ReceiverStream::new(rcv));