core: Enforce single shared database object per database file

We need to ensures that there is a single, shared `Database` object per
a database file. We need because it is not safe to have multiple
independent WAL files open because coordination happens at process-level
POSIX file advisory locks.

Fixes #2267
Co-authored-by: ultraman <sunhuayangak47@gmail.com>

core/lib.rs

@@ -72,7 +72,7 @@ use std::{
     num::NonZero,
     ops::Deref,
     rc::Rc,
-    sync::{Arc, Mutex},
+    sync::{Arc, LazyLock, Mutex, Weak},
 };
 #[cfg(feature = "fs")]
 use storage::database::DatabaseFile;
@@ -108,6 +108,15 @@ pub(crate) type MvStore = mvcc::MvStore<mvcc::LocalClock>;
 
 pub(crate) type MvCursor = mvcc::cursor::ScanCursor<mvcc::LocalClock>;
 
+/// The database manager ensures that there is a single, shared
+/// `Database` object per a database file. We need because it is not safe
+/// to have multiple independent WAL files open because coordination
+/// happens at process-level POSIX file advisory locks.
+static DATABASE_MANAGER: LazyLock<Mutex<HashMap<String, Weak<Database>>>> =
+    LazyLock::new(|| Mutex::new(HashMap::new()));
+
+/// The `Database` object contains per database file state that is shared
+/// between multiple connections.
 pub struct Database {
     mv_store: Option<Rc<MvStore>>,
     schema: Mutex<Arc<Schema>>,
@@ -224,6 +233,36 @@ impl Database {
         flags: OpenFlags,
         enable_mvcc: bool,
         enable_indexes: bool,
+    ) -> Result<Arc<Database>> {
+        if path == ":memory:" {
+            return Self::do_open_with_flags(io, path, db_file, flags, enable_mvcc, enable_indexes);
+        }
+
+        let mut registry = DATABASE_MANAGER.lock().unwrap();
+
+        let canonical_path = std::fs::canonicalize(path)
+            .ok()
+            .and_then(|p| p.to_str().map(|s| s.to_string()))
+            .unwrap_or_else(|| path.to_string());
+
+        if let Some(db) = registry.get(&canonical_path) {
+            if let Some(db) = db.upgrade() {
+                return Ok(db);
+            }
+        }
+        let db = Self::do_open_with_flags(io, path, db_file, flags, enable_mvcc, enable_indexes)?;
+        registry.insert(canonical_path, Arc::downgrade(&db));
+        Ok(db)
+    }
+
+    #[allow(clippy::arc_with_non_send_sync)]
+    fn do_open_with_flags(
+        io: Arc<dyn IO>,
+        path: &str,
+        db_file: Arc<dyn DatabaseStorage>,
+        flags: OpenFlags,
+        enable_mvcc: bool,
+        enable_indexes: bool,
     ) -> Result<Arc<Database>> {
         let wal_path = format!("{path}-wal");
         let maybe_shared_wal = WalFileShared::open_shared_if_exists(&io, wal_path.as_str())?;