diff --git a/datasette/database.py b/datasette/database.py
index 32e1d19d..73687c51 100644
--- a/datasette/database.py
+++ b/datasette/database.py
@@ -223,7 +223,11 @@ class Database:
         else:
             return "db"
 
-    def connect(self, write=False):
+    def connect(self, write=False, track=True):
+        """Open a new connection. ``track=False`` leaves it out of
+        ``_all_connections``: the caller closes it itself and close() must
+        never close it from another thread while it is in use (the
+        SchemaWatcher's short-lived scan connections)."""
         extra_kwargs = {}
         if write:
             extra_kwargs["isolation_level"] = "IMMEDIATE"
@@ -234,11 +238,13 @@ class Database:
             )
             if not write:
                 conn.execute("PRAGMA query_only=1")
-            self._all_connections.append(conn)
+            if track:
+                self._all_connections.append(conn)
             return conn
         if self.is_memory:
             conn = sqlite3.connect(":memory:", uri=True, check_same_thread=False)
-            self._all_connections.append(conn)
+            if track:
+                self._all_connections.append(conn)
             return conn
 
         # mode=ro or immutable=1?
@@ -261,7 +267,8 @@ class Database:
         conn = sqlite3.connect(
             f"file:{self.path}{qs}", uri=True, check_same_thread=False, **extra_kwargs
         )
-        self._all_connections.append(conn)
+        if track:
+            self._all_connections.append(conn)
         if self.is_temp_disk and not self._wal_enabled:
             conn.execute("PRAGMA journal_mode=WAL")
             self._wal_enabled = True
diff --git a/datasette/schema_watcher.py b/datasette/schema_watcher.py
index 6fa5ed7f..d3f84e92 100644
--- a/datasette/schema_watcher.py
+++ b/datasette/schema_watcher.py
@@ -605,26 +605,23 @@ class SchemaWatcher:
     # scanning + storing
     # ------------------------------------------------------------------
     def _connect(self, db, prepare):
-        conn = db.connect()
+        # Untracked: Database.close() on the event loop thread must not close
+        # this connection while a worker thread is using it (that segfaults).
+        # It is always closed by _close() in the same thread that opened it.
+        conn = db.connect(track=False)
         try:
             if prepare:
                 self.ds._prepare_connection(conn, db.name)
             else:
                 conn.row_factory = None
         except Exception:
-            self._close(db, conn)
+            conn.close()
             raise
         return conn
 
     @staticmethod
     def _close(db, conn):
-        try:
-            conn.close()
-        finally:
-            try:
-                db._all_connections.remove(conn)
-            except ValueError:
-                pass
+        conn.close()
 
     def _scan_sync(self, state):
         """Read schema_version + full schema in one read transaction."""
diff --git a/tests/test_schema_watcher.py b/tests/test_schema_watcher.py
index 2b876809..df9f1a73 100644
--- a/tests/test_schema_watcher.py
+++ b/tests/test_schema_watcher.py
@@ -550,3 +550,30 @@ async def test_persisted_fingerprints_skip_unchanged_files(tmp_path, monkeypatch
     assert await catalog_tables(ds2, "p1") == ["added_offline", "t"]
     assert await catalog_tables(ds2, "p0") == ["t"]
     ds2.close()
+
+
+@pytest.mark.asyncio
+async def test_remove_database_during_catalog_scan(tmp_path):
+    # remove_database() closes the Database while the watcher may still be
+    # scanning it in a worker thread. The scan connection is not tracked by
+    # the Database, so close() must not pull it out from under the scan
+    # (that used to segfault).
+    path = str(tmp_path / "big.db")
+    conn = sqlite3.connect(path)
+    for i in range(150):
+        conn.execute(f"create table t{i} (a, b, c integer references t0(a))")
+        conn.execute(f"create index i{i} on t{i}(a)")
+    conn.commit()
+    conn.close()
+    ds = Datasette(settings={"schema_watch_interval_ms": 0})
+    await ds.invoke_startup()
+    for i in range(60):
+        ds.add_database(Database(ds, path=path, is_mutable=i % 2 == 0), name=f"x{i}")
+        await asyncio.sleep(0.0005 * (i % 5))
+        ds.remove_database(f"x{i}")
+    await asyncio.sleep(0.2)
+    rows = await ds.get_internal_database().execute(
+        "select count(*) from catalog_tables where database_name like 'x%'"
+    )
+    assert rows.first()[0] == 0
+    ds.close()