Merge pull request #1750 from tursodatabase/lucio/clean-up-schema-rep…

…lica

sqld: schema db and replica fixes

libsql-server/src/rpc/proxy.rs

@@ -635,7 +635,7 @@ impl Proxy for ProxyService {
                         Ok(conn) => {
                             if !conn.is_primary() {
                                 return Err(tonic::Status::failed_precondition(
-                                    "cannot run schema migration against a replica",
+                                    "cannot run schema migration against a replica from a replica",
                                 ));
                             }
 
@@ -710,7 +710,7 @@ impl Proxy for ProxyService {
                     Ok(conn) => {
                         if !conn.is_primary() {
                             return Err(tonic::Status::failed_precondition(
-                                "cannot run schema migration against a replica",
+                                "cannot run schema migration against a replica from a replica",
                             ));
                         }
 

libsql-server/src/rpc/replication/replication_log.rs

@@ -351,6 +351,14 @@ impl ReplicationLog for ReplicationLogService {
         let (logger, config, version, _, _) =
             self.logger_from_namespace(namespace, &req, false).await?;
 
+        // If we are a shared schema and serving externally (aka to embedded replica's) then
+        // return an error.
+        if config.is_shared_schema && !self.service_internal {
+            return Err(tonic::Status::failed_precondition(
+                "cannot replicate a shared schema db",
+            ));
+        }
+
         let session_hash = self.encode_session_token(version);
 
         let response = HelloResponse {

libsql-server/tests/cluster/mod.rs

@@ -311,3 +311,32 @@ fn large_proxy_query() {
 
     sim.run().unwrap();
 }
+
+#[test]
+fn replicate_from_shared_schema() {
+    let mut sim = Builder::new()
+        .simulation_duration(Duration::from_secs(10000))
+        .tcp_capacity(100000)
+        .build();
+    make_cluster(&mut sim, 1, true);
+
+    sim.client("client", async {
+        let db = Database::open_remote_with_connector("http://primary:8080", "", TurmoilConnector)
+            .unwrap();
+        let conn = db.connect().unwrap();
+
+        conn.execute("create table test (x)", ()).await.unwrap();
+
+        let db = Database::open_remote_with_connector("http://replica0:8080", "", TurmoilConnector)
+            .unwrap();
+        let conn = db.connect().unwrap();
+
+        conn.execute_batch("select * from sqlite_master;")
+            .await
+            .unwrap();
+
+        Ok(())
+    });
+
+    sim.run().unwrap();
+}

libsql-server/tests/cluster/schema_dbs.rs

@@ -129,3 +129,64 @@ fn schema_migration_basics() {
 
     sim.run().unwrap();
 }
+
+#[test]
+fn schema_migration_via_replica() {
+    let mut sim = Builder::new()
+        .simulation_duration(Duration::from_secs(1000))
+        .build();
+    make_cluster(&mut sim, 1, true);
+
+    sim.client("client", async {
+        let http = Client::new();
+
+        assert!(http
+            .post(
+                "http://primary:9090/v1/namespaces/schema/create",
+                json!({ "shared_schema": true })
+            )
+            .await
+            .unwrap()
+            .status()
+            .is_success());
+        assert!(http
+            .post(
+                "http://primary:9090/v1/namespaces/foo/create",
+                json!({ "shared_schema_name": "schema" })
+            )
+            .await
+            .unwrap()
+            .status()
+            .is_success());
+
+        {
+            let db = Database::open_remote_with_connector(
+                "http://schema.primary:8080",
+                "",
+                TurmoilConnector,
+            )
+            .unwrap();
+            let conn = db.connect().unwrap();
+            conn.execute("create table test (x)", ()).await.unwrap();
+        }
+
+        {
+            let db = Database::open_remote_with_connector(
+                "http://schema.replica0:8080",
+                "",
+                TurmoilConnector,
+            )
+            .unwrap();
+            let conn = db.connect().unwrap();
+            conn.execute("select * from sqlite_master;", ())
+                .await
+                .unwrap();
+
+            conn.execute("create table foo (x)", ()).await.unwrap();
+        }
+
+        Ok(())
+    });
+
+    sim.run().unwrap();
+}

libsql-server/tests/embedded_replica/mod.rs

@@ -1637,3 +1637,62 @@ fn replicated_synced_frames_zero_when_no_data_synced() {
 
     sim.run().unwrap();
 }
+
+#[test]
+fn schema_db() {
+    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();
+
+    let mut sim = Builder::new()
+        .simulation_duration(Duration::from_secs(1000))
+        .build();
+
+    make_primary(&mut sim, tmp_host_path.clone());
+
+    sim.client("client", async move {
+        let http = Client::new();
+
+        assert!(http
+            .post(
+                "http://primary:9090/v1/namespaces/schema/create",
+                json!({ "shared_schema": true })
+            )
+            .await
+            .unwrap()
+            .status()
+            .is_success());
+        assert!(http
+            .post(
+                "http://primary:9090/v1/namespaces/foo/create",
+                json!({ "shared_schema_name": "schema" })
+            )
+            .await
+            .unwrap()
+            .status()
+            .is_success());
+
+        let path = tmp_embedded_path.join("embedded");
+        let db = Database::open_with_remote_sync_connector(
+            path.to_str().unwrap(),
+            "http://schema.primary:8080",
+            "",
+            TurmoilConnector,
+            false,
+            None,
+        )
+        .await
+        .unwrap();
+
+        db.sync().await.unwrap_err();
+
+        let conn = db.connect().unwrap();
+
+        conn.execute("create table test (x)", ()).await.unwrap_err();
+
+        Ok(())
+    });
+
+    sim.run().unwrap();
+}