store: Guard against a race setting up the 'sharded' namespace

Author: lutter

store/postgres/src/connection_pool.rs

@@ -35,7 +35,7 @@ use postgres::config::{Config, Host};
 
 use crate::advisory_lock::with_migration_lock;
 use crate::catalog;
-use crate::primary::{self, NAMESPACE_PUBLIC};
+use crate::primary::{self, Namespace, NAMESPACE_PUBLIC};
 use crate::{Shard, PRIMARY_SHARD};
 
 /// Tables that we map from the primary into `primary_public` in each shard
@@ -1225,30 +1225,38 @@ impl PoolInner {
             return Ok(());
         }
 
-        info!(&self.logger, "Creating cross-shard views");
         let mut conn = self.get()?;
+        let sharded = Namespace::special(ForeignServer::CROSS_SHARD_NSP);
+        if catalog::has_namespace(&mut conn, &sharded)? {
+            // We dropped the namespace before, but another node must have
+            // recreated it in the meantime so we don't need to do anything
+            return Ok(());
+        }
 
-        conn.transaction(|conn| {
-            let query = format!(
-                "create schema if not exists {}",
-                ForeignServer::CROSS_SHARD_NSP
-            );
-            conn.batch_execute(&query)?;
-            for (src_nsp, src_tables) in SHARDED_TABLES {
-                // Pairs of (shard, nsp) for all servers
-                let nsps = shard_nsp_pairs(&self.shard, src_nsp, servers);
-                for src_table in src_tables {
-                    let create_view = catalog::create_cross_shard_view(
-                        conn,
-                        src_nsp,
-                        src_table,
-                        ForeignServer::CROSS_SHARD_NSP,
-                        &nsps,
-                    )?;
-                    conn.batch_execute(&create_view)?;
-                }
+        info!(&self.logger, "Creating cross-shard views");
+        with_migration_lock(&mut conn, |conn| {
+            if catalog::has_namespace(conn, &sharded)? {
+                return Ok(());
             }
-            Ok(())
+            conn.transaction(|conn| {
+                let query = format!("create schema {}", ForeignServer::CROSS_SHARD_NSP);
+                conn.batch_execute(&query)?;
+                for (src_nsp, src_tables) in SHARDED_TABLES {
+                    // Pairs of (shard, nsp) for all servers
+                    let nsps = shard_nsp_pairs(&self.shard, src_nsp, servers);
+                    for src_table in src_tables {
+                        let create_view = catalog::create_cross_shard_view(
+                            conn,
+                            src_nsp,
+                            src_table,
+                            ForeignServer::CROSS_SHARD_NSP,
+                            &nsps,
+                        )?;
+                        conn.batch_execute(&create_view)?;
+                    }
+                }
+                Ok(())
+            })
         })
     }