Compare commits

..

4 commits

Author SHA1 Message Date
Zain Dana Harper
7403ae68bb Give each non-blocking write a distinct task id, refs #2860, #2859
execute_write_fn(fn, block=False) is documented to return "a UUID representing
the queued task". Two things stopped that being true.

_send_to_write_thread() derived the id from uuid.uuid5(NAMESPACE_DNS,
"datasette.io"), which is deterministic, so every non-blocking write in every
database in every process returned 3f143baa-4e3d-5842-a36f-4fa2f683b72f. A
constant cannot identify a particular task. Now uuid4(). Refs #2860.

With num_sql_threads=0 there is no write thread, so execute_write_fn took the
synchronous branch and `result` was the write function's return value, normally
None. The block=False path then unpacked it unconditionally and raised
TypeError: cannot unpack non-iterable NoneType object. The non-threaded branch
now returns the same (task_id, reply_future) shape, with the future already
resolved because the write has finished, so both modes share one code path.
Refs #2859.

test_execute_write_fn_block_false only asserted isinstance(task_id, uuid.UUID),
which a constant satisfies. The new test is parametrized over threaded and
non-threaded and asserts two calls return different ids, so either regression
fails it.
2026-09-08 10:03:52 -07:00
Simon Willison
bdc9731740
check-latest: true, add 3.15 to test matrix, to test RCs (#2895)
See https://simonwillison.net/2026/Sep/1/python-315-rc-2/
2026-09-01 13:37:15 -07:00
Alex Garcia
3e018bb1b5
Run startup via ASGI lifespan instead of waiting for the first request (#2887)
* Run startup via ASGI lifespan instead of waiting for the first request
* Ensure immutable table counts still precompute when startup ran first
2026-09-01 09:39:25 -07:00
Alex Garcia
e78b8a2e6a
Run datasette serve startup and uvicorn on a single event loop (#2886)
* Run datasette serve startup and uvicorn on a single event loop
* Move the serve-subprocess test plumbing into a conftest fixture
* Fix datasette-litestream URL and trim marker-task test comments
* Explain why serve_with_plugins needs a subprocess and plugin files
* Apply ruff 0.16 and black fixes
* Tweaked some comments
2026-09-01 09:32:37 -07:00
8 changed files with 379 additions and 31 deletions

View file

@ -11,16 +11,17 @@ jobs:
strategy: strategy:
fail-fast: false fail-fast: false
matrix: matrix:
python-version: ["3.10", "3.11", "3.12", "3.13", "3.14"] python-version: ["3.10", "3.11", "3.12", "3.13", "3.14", "3.15"]
steps: steps:
- uses: actions/checkout@v7 - uses: actions/checkout@v7
- name: Set up Python ${{ matrix.python-version }} - name: Set up Python ${{ matrix.python-version }}
uses: actions/setup-python@v6 uses: actions/setup-python@v7
with: with:
python-version: ${{ matrix.python-version }} python-version: ${{ matrix.python-version }}
allow-prereleases: true allow-prereleases: true
cache: pip cache: pip
cache-dependency-path: pyproject.toml cache-dependency-path: pyproject.toml
check-latest: true
- name: Build extension for --load-extension test - name: Build extension for --load-extension test
run: |- run: |-
(cd tests && gcc ext.c -fPIC -shared -o ext.so) (cd tests && gcc ext.c -fPIC -shared -o ext.so)

View file

@ -453,8 +453,10 @@ class Datasette:
self.databases = collections.OrderedDict() self.databases = collections.OrderedDict()
self.actions = {} # .invoke_startup() will populate this self.actions = {} # .invoke_startup() will populate this
self._column_types = {} # .invoke_startup() will populate this self._column_types = {} # .invoke_startup() will populate this
self._setup_db_done = False
try: try:
self._refresh_schemas_lock = asyncio.Lock() self._refresh_schemas_lock = asyncio.Lock()
self._startup_lock = asyncio.Lock()
except RuntimeError as rex: except RuntimeError as rex:
# Workaround for intermittent test failure, see: # Workaround for intermittent test failure, see:
# https://github.com/simonw/datasette/issues/1802 # https://github.com/simonw/datasette/issues/1802
@ -462,6 +464,7 @@ class Datasette:
loop = asyncio.new_event_loop() loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop) asyncio.set_event_loop(loop)
self._refresh_schemas_lock = asyncio.Lock() self._refresh_schemas_lock = asyncio.Lock()
self._startup_lock = asyncio.Lock()
else: else:
raise raise
self.crossdb = crossdb self.crossdb = crossdb
@ -2803,24 +2806,52 @@ class Datasette:
raise RowNotFound(db.name, table_name, pk_values) raise RowNotFound(db.name, table_name, pk_values)
return ResolvedRow(db, table_name, sql, params, pks, pk_values, results.first()) return ResolvedRow(db, table_name, sql, params, pks, pk_values, results.first())
async def _startup_sequence(self):
"""Idempotently run the full startup sequence: table counts for
immutable databases, then invoke_startup(). Safe to call more than
once and safe to call concurrently - callers block until whichever
call got there first has finished.
This is the single entry point used by both AsgiLifespan (so
real deployments finish startup before accepting requests) and
AsgiRunOnFirstRequest (the fallback for hosts that never send
lifespan events, e.g. DatasetteClient's httpx.ASGITransport), and
`datasette serve` (cli.py) calls it too. The fast path below checks
both `_startup_invoked` and `_setup_db_done` - not just the former -
so that a bare `await ds.invoke_startup()` made by a caller ahead of
`_startup_sequence()` (which only sets `_startup_invoked`) can't
make this method skip the immutable-database table-count precompute.
"""
if self._startup_invoked and self._setup_db_done:
return
async with self._startup_lock:
if self._startup_invoked and self._setup_db_done:
return
if not self._setup_db_done:
# First time server starts up, calculate table counts for
# immutable databases
for database in self.databases.values():
if not database.is_mutable:
await database.table_counts(limit=60 * 60 * 1000)
self._setup_db_done = True
await self.invoke_startup()
def app(self): def app(self):
"""Returns an ASGI app function that serves the whole of Datasette""" """Returns an ASGI app function that serves the whole of Datasette"""
routes = self._routes() routes = self._routes()
async def setup_db():
# First time server starts up, calculate table counts for immutable databases
for database in self.databases.values():
if not database.is_mutable:
await database.table_counts(limit=60 * 60 * 1000)
async def _close_on_shutdown(): async def _close_on_shutdown():
self.close() self.close()
asgi = CrossOriginProtectionMiddleware(DatasetteRouter(self, routes), self) asgi = CrossOriginProtectionMiddleware(DatasetteRouter(self, routes), self)
if self.setting("trace_debug"): if self.setting("trace_debug"):
asgi = AsgiTracer(asgi) asgi = AsgiTracer(asgi)
asgi = AsgiLifespan(asgi, on_shutdown=[_close_on_shutdown]) asgi = AsgiLifespan(
asgi = AsgiRunOnFirstRequest(asgi, on_startup=[setup_db, self.invoke_startup]) asgi,
on_startup=[self._startup_sequence],
on_shutdown=[_close_on_shutdown],
)
asgi = AsgiRunOnFirstRequest(asgi, on_startup=[self._startup_sequence])
for wrapper in pm.hook.asgi_wrapper(datasette=self): for wrapper in pm.hook.asgi_wrapper(datasette=self):
asgi = wrapper(asgi) asgi = wrapper(asgi)
return asgi return asgi

View file

@ -670,11 +670,9 @@ def serve(
raise click.ClickException("--token can only be used with --get") raise click.ClickException("--token can only be used with --get")
if get: if get:
# Run async soundness checks before startup hooks, since invoke_startup # --get means we don't run Uvicorn at all
# now populates internal tables which requires querying each database
run_sync(lambda: check_databases(ds)) run_sync(lambda: check_databases(ds))
# Run the "startup" plugin hooks
try: try:
run_sync(ds.invoke_startup) run_sync(ds.invoke_startup)
except StartupError as e: except StartupError as e:
@ -709,13 +707,15 @@ def serve(
# on the loop (asyncio.create_task, Lock/Queue/Event objects, ...) is # on the loop (asyncio.create_task, Lock/Queue/Event objects, ...) is
# still alive when the server starts handling requests. # still alive when the server starts handling requests.
async def _serve_async(): async def _serve_async():
# Run async soundness checks before startup hooks, since invoke_startup # Populate internal catalog tables before invoke_startup
# now populates internal tables which requires querying each database
await check_databases(ds) await check_databases(ds)
# Run the "startup" plugin hooks # Run the full startup sequence (immutable-database table-count
# precompute + the "startup" plugin hooks) via the same entry point
# AsgiLifespan/AsgiRunOnFirstRequest use, so it's not skipped when
# uvicorn's lifespan.startup fires moments later.
try: try:
await ds.invoke_startup() await ds._startup_sequence()
except StartupError as e: except StartupError as e:
raise click.ClickException(e.args[0]) raise click.ClickException(e.args[0])

View file

@ -354,6 +354,15 @@ class Database:
result = fn(self._write_connection) result = fn(self._write_connection)
else: else:
result = fn(self._write_connection) result = fn(self._write_connection)
if not block:
# There is no write thread here, so the write has already
# finished. Hand back the same (task_id, reply_future) shape
# _send_to_write_thread() returns, with the future already
# resolved, so the block=False path below is identical in
# both modes.
reply_future = asyncio.get_running_loop().create_future()
reply_future.set_result(result)
result = (uuid.uuid4(), reply_future)
else: else:
result = await self._send_to_write_thread( result = await self._send_to_write_thread(
fn, block=block, transaction=transaction fn, block=block, transaction=transaction
@ -425,7 +434,7 @@ class Database:
) )
self._write_thread.name = f"_execute_writes for database {self.name}" self._write_thread.name = f"_execute_writes for database {self.name}"
self._write_thread.start() self._write_thread.start()
task_id = uuid.uuid5(uuid.NAMESPACE_DNS, "datasette.io") task_id = uuid.uuid4()
loop = asyncio.get_running_loop() loop = asyncio.get_running_loop()
reply_future = loop.create_future() reply_future = loop.create_future()
self._write_queue.put( self._write_queue.put(

View file

@ -1,3 +1,4 @@
import asyncio
import json import json
import re import re
from http.cookies import Morsel, SimpleCookie from http.cookies import Morsel, SimpleCookie
@ -300,12 +301,24 @@ class AsgiLifespan:
while True: while True:
message = await receive() message = await receive()
if message["type"] == "lifespan.startup": if message["type"] == "lifespan.startup":
for fn in self.on_startup: try:
await fn() for fn in self.on_startup:
await fn()
except Exception as e: # noqa: BLE001
await send(
{"type": "lifespan.startup.failed", "message": str(e)}
)
return
await send({"type": "lifespan.startup.complete"}) await send({"type": "lifespan.startup.complete"})
elif message["type"] == "lifespan.shutdown": elif message["type"] == "lifespan.shutdown":
for fn in self.on_shutdown: try:
await fn() for fn in self.on_shutdown:
await fn()
except Exception as e: # noqa: BLE001
await send(
{"type": "lifespan.shutdown.failed", "message": str(e)}
)
return
await send({"type": "lifespan.shutdown.complete"}) await send({"type": "lifespan.shutdown.complete"})
return return
else: else:
@ -624,10 +637,23 @@ class AsgiRunOnFirstRequest:
self.asgi = asgi self.asgi = asgi
self.on_startup = on_startup self.on_startup = on_startup
self._started = False self._started = False
# Guards against concurrent early requests interleaving with startup:
# without this, several requests could all observe `_started is
# False` and proceed before any of them finish running the hooks.
self._lock = asyncio.Lock()
async def __call__(self, scope, receive, send): async def __call__(self, scope, receive, send):
if not self._started: # Leave "lifespan" scope events alone - this shim only exists as a
self._started = True # fallback for hosts that never send them. It wraps AsgiLifespan, so
for hook in self.on_startup: # if it ran on_startup here too, a startup exception would escape
await hook() # before AsgiLifespan's own try/except got a chance to turn it into
# a lifespan.startup.failed message.
if scope["type"] != "lifespan" and not self._started:
async with self._lock:
# Re-check: another request may have finished startup while
# we were waiting for the lock.
if not self._started:
for hook in self.on_startup:
await hook()
self._started = True
return await self.asgi(scope, receive, send) return await self.asgi(scope, receive, send)

View file

@ -127,11 +127,6 @@ def test_startup_error_fails_fast_before_port_binds(serve_with_plugins):
A "startup" plugin hook that raises StartupError must fail fast: print A "startup" plugin hook that raises StartupError must fail fast: print
the message, exit non-zero, and never accept a connection on the port - the message, exit non-zero, and never accept a connection on the port -
the failure must happen before uvicorn.Server binds the socket. the failure must happen before uvicorn.Server binds the socket.
Note this is a characterization test, not a regression test: it also
passes on unmodified main, where startup already ran ahead of
uvicorn.run(). It earns its keep once startup moves into the ASGI
lifespan, where fail-fast is genuinely at risk.
""" """
proc, port = serve_with_plugins( proc, port = serve_with_plugins(
{"startup_error_plugin": STARTUP_ERROR_PLUGIN}, wait_for_startup=False {"startup_error_plugin": STARTUP_ERROR_PLUGIN}, wait_for_startup=False

View file

@ -705,6 +705,33 @@ async def test_execute_write_fn_block_false(db):
assert isinstance(task_id, uuid.UUID) assert isinstance(task_id, uuid.UUID)
@pytest.mark.asyncio
@pytest.mark.parametrize("disable_threads", (False, True))
async def test_execute_write_fn_block_false_returns_uuid(tmp_path, disable_threads):
# block=False is documented to return "a UUID representing the queued task".
# With num_sql_threads=0 there is no write thread, so the non-threaded branch
# has to satisfy the same contract as the threaded one.
settings = {"num_sql_threads": 0} if disable_threads else {}
ds = Datasette([], memory=True, settings=settings)
await ds.invoke_startup()
db = ds.add_memory_database("test_block_false")
await db.execute_write(
"create table if not exists t (id integer primary key, v text)"
)
def write_fn(conn):
conn.execute("insert into t (v) values ('a')")
# Returns None, like most write functions.
task_id = await db.execute_write_fn(write_fn, block=False)
assert isinstance(task_id, uuid.UUID)
# Distinct per call, so a caller can tell two queued tasks apart.
second = await db.execute_write_fn(write_fn, block=False)
assert isinstance(second, uuid.UUID)
assert second != task_id
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_execute_write_fn_block_true(db): async def test_execute_write_fn_block_true(db):
def write_fn(conn): def write_fn(conn):

259
tests/test_lifespan.py Normal file
View file

@ -0,0 +1,259 @@
"""
Tests for wiring Datasette startup (setup_db table counts + invoke_startup)
into the ASGI lifespan protocol.
These exercise Datasette._startup_sequence() via three different callers:
- AsgiLifespan, by hand-driving lifespan.startup messages (no HTTP request)
- AsgiRunOnFirstRequest, the fallback for hosts that never send lifespan
events (this is what DatasetteClient / plain httpx.ASGITransport uses)
- Both at once, to prove startup hooks run at most once
"""
import asyncio
import contextlib
import sqlite3
import httpx
import pytest
from datasette import hookimpl
from datasette.app import Datasette
from datasette.database import Database
from datasette.plugins import pm
async def _drive_lifespan_startup(app):
"""Send a single lifespan.startup message into app's ASGI lifespan loop
and return the list of messages sent back - without ever sending
lifespan.shutdown. Mirrors what a real server does: after startup
completes it parks waiting for the next event. We cancel that wait
once we've observed the startup response, rather than closing the
Datasette instance down with a shutdown message.
"""
messages_sent = []
startup_responded = asyncio.Event()
delivered = False
async def receive():
nonlocal delivered
if not delivered:
delivered = True
return {"type": "lifespan.startup"}
# No further messages: block until the task is cancelled below,
# same as a real server parked waiting for lifespan.shutdown.
await asyncio.Event().wait()
async def send(message):
messages_sent.append(message)
startup_responded.set()
task = asyncio.create_task(app({"type": "lifespan"}, receive, send))
try:
await asyncio.wait_for(startup_responded.wait(), timeout=5)
finally:
task.cancel()
with contextlib.suppress(asyncio.CancelledError):
await task
return messages_sent
@pytest.mark.asyncio
async def test_lifespan_startup_runs_before_any_request():
ds = Datasette(memory=True)
assert ds._startup_invoked is False
app = ds.app()
messages = await _drive_lifespan_startup(app)
assert {"type": "lifespan.startup.complete"} in messages
assert ds._startup_invoked is True
# Internal catalog tables should be populated too, entirely without an
# HTTP request having been made.
internal_db = ds.get_internal_database()
databases = await internal_db.execute("select * from catalog_databases")
assert len(databases.rows) >= 1
@pytest.mark.asyncio
async def test_lifespan_startup_failure_reports_lifespan_startup_failed():
class RaisingStartupPlugin:
__name__ = "RaisingStartupPlugin"
@hookimpl
def startup(self, datasette):
async def inner():
raise RuntimeError("boom from startup hook")
return inner
ds = Datasette(memory=True)
pm.register(RaisingStartupPlugin(), name="raising_startup_plugin")
try:
app = ds.app()
messages = await _drive_lifespan_startup(app)
finally:
pm.unregister(name="raising_startup_plugin")
assert messages == [
{"type": "lifespan.startup.failed", "message": "boom from startup hook"}
]
# The exception happened before invoke_startup() got to the end of its
# body, so startup is not considered to have completed.
assert ds._startup_invoked is False
@pytest.mark.asyncio
async def test_startup_runs_exactly_once_across_lifespan_and_first_request():
call_count = {"n": 0}
class CountingStartupPlugin:
__name__ = "CountingStartupPlugin"
@hookimpl
def startup(self, datasette):
async def inner():
call_count["n"] += 1
return inner
ds = Datasette(memory=True)
pm.register(CountingStartupPlugin(), name="counting_startup_plugin")
try:
# Build the ASGI app once, the way a real deployment does - and
# reuse the SAME app instance for both the lifespan drive and the
# HTTP requests below, since a fresh ds.app() call would reset the
# AsgiRunOnFirstRequest fallback's state.
app = ds.app()
messages = await _drive_lifespan_startup(app)
assert {"type": "lifespan.startup.complete"} in messages
assert call_count["n"] == 1
# A first HTTP request (as if the host never sent lifespan events,
# or lifespan already ran) should not run the hook again.
transport = httpx.ASGITransport(app=app)
async with httpx.AsyncClient(
transport=transport, base_url="http://localhost"
) as client:
response1 = await client.get("/-/versions.json")
assert response1.status_code == 200
# ... nor should a second, repeat request.
response2 = await client.get("/-/versions.json")
assert response2.status_code == 200
finally:
pm.unregister(name="counting_startup_plugin")
assert call_count["n"] == 1
@pytest.mark.asyncio
async def test_no_lifespan_first_request_still_triggers_startup():
# Pin today's behavior: a client that never drives ASGI lifespan events
# at all (like httpx.ASGITransport, which DatasetteClient uses) still
# gets startup armed by the AsgiRunOnFirstRequest fallback.
ds = Datasette(memory=True)
assert ds._startup_invoked is False
app = ds.app()
transport = httpx.ASGITransport(app=app)
async with httpx.AsyncClient(
transport=transport, base_url="http://localhost"
) as client:
response = await client.get("/-/versions.json")
assert response.status_code == 200
assert ds._startup_invoked is True
internal_db = ds.get_internal_database()
databases = await internal_db.execute("select * from catalog_databases")
assert len(databases.rows) >= 1
@pytest.mark.asyncio
async def test_datasette_client_first_request_triggers_startup():
# Same as above, but through the real DatasetteClient (ds.client) that
# plugins and tests actually use, to confirm nothing regressed there.
ds = Datasette(memory=True)
assert ds._startup_invoked is False
response = await ds.client.get("/-/versions.json")
assert response.status_code == 200
assert ds._startup_invoked is True
@pytest.mark.asyncio
async def test_concurrent_first_requests_all_wait_for_slow_startup():
call_count = {"n": 0}
class SlowStartupPlugin:
__name__ = "SlowStartupPlugin"
@hookimpl
def startup(self, datasette):
async def inner():
call_count["n"] += 1
await asyncio.sleep(0.2)
return inner
ds = Datasette(memory=True)
pm.register(SlowStartupPlugin(), name="slow_startup_plugin")
try:
app = ds.app()
transport = httpx.ASGITransport(app=app)
async with httpx.AsyncClient(
transport=transport, base_url="http://localhost"
) as client:
responses = await asyncio.gather(
*[client.get("/-/versions.json") for _ in range(10)]
)
finally:
pm.unregister(name="slow_startup_plugin")
# Every one of the 10 simultaneous first requests must have blocked
# until startup actually finished, not raced ahead of it.
assert all(response.status_code == 200 for response in responses)
assert call_count["n"] == 1
assert ds._startup_invoked is True
@pytest.mark.asyncio
async def test_setup_db_still_runs_when_invoke_startup_ran_first(tmp_path, monkeypatch):
# Regression test: `datasette serve` (cli.py _serve_async) calls
# ds.invoke_startup() directly, before uvicorn ever sends a
# lifespan.startup event that drives _startup_sequence(). If
# _startup_sequence()'s fast path only checked `_startup_invoked`, it
# would see startup already done and skip the immutable-database
# table-count precompute (setup_db) entirely - a silent regression
# versus main, where AsgiRunOnFirstRequest ran setup_db unconditionally
# on request #1.
db_path = tmp_path / "immutable.db"
conn = sqlite3.connect(str(db_path))
conn.execute("create table t (id integer primary key)")
conn.commit()
conn.close()
ds = Datasette([], immutables=[str(db_path)])
call_count = {"n": 0}
original_table_counts = Database.table_counts
async def counting_table_counts(self, *args, **kwargs):
call_count["n"] += 1
return await original_table_counts(self, *args, **kwargs)
monkeypatch.setattr(Database, "table_counts", counting_table_counts)
# Simulate the CLI path: invoke_startup() runs directly and completes
# BEFORE _startup_sequence() ever gets a chance to run setup_db.
await ds.invoke_startup()
assert ds._startup_invoked is True
assert call_count["n"] == 0
# The lifespan/first-request path (or the CLI itself, per the fix)
# calling the shared entry point afterwards must still precompute
# table counts for immutable databases.
await ds._startup_sequence()
assert call_count["n"] == 1
assert ds._setup_db_done is True
# Idempotency: a second call must not recompute.
await ds._startup_sequence()
assert call_count["n"] == 1