mirror of
https://github.com/simonw/datasette.git
synced 2026-10-19 07:47:00 +02:00
Schema watcher: scan connections are untracked so close() cannot pull them from under a scan
remove_database() closes the Database, which closed every connection in _all_connections - including a catalog scan connection still in use on a worker thread: segfault (crashed an xdist worker in the full suite, reproduced with an add/remove stress loop). Database.connect(track=False) leaves the watcher's short-lived connections to the watcher, which always closes them on the thread that opened them. Regression test added. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01EwkLo2rJWidpkDVT3HkvNv
This commit is contained in:
parent
a7359a7065
commit
c87eeceb48
3 changed files with 44 additions and 13 deletions
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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."""
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue