From cd9daa266c574dd877fec737010fb364db80c432 Mon Sep 17 00:00:00 2001
From: Alex Garcia <alexsebastian.garcia@gmail.com>
Date: Thu, 30 Jul 2026 17:25:55 -0700
Subject: [PATCH 1/8] Run datasette serve startup and uvicorn on a single event
 loop

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 datasette/cli.py               |  94 ++++++++++-------
 pyproject.toml                 |   2 +-
 tests/test_cli_serve_server.py | 181 +++++++++++++++++++++++++++++++++
 3 files changed, 238 insertions(+), 39 deletions(-)

diff --git a/datasette/cli.py b/datasette/cli.py
index 57db83b6..06fa6199 100644
--- a/datasette/cli.py
+++ b/datasette/cli.py
@@ -663,16 +663,6 @@ def serve(
         # Private utility mechanism for writing unit tests
         return ds
 
-    # Run async soundness checks before startup hooks, since invoke_startup
-    # now populates internal tables which requires querying each database
-    run_sync(lambda: check_databases(ds))
-
-    # Run the "startup" plugin hooks
-    try:
-        run_sync(ds.invoke_startup)
-    except StartupError as e:
-        raise click.ClickException(e.args[0])
-
     if headers and not get:
         raise click.ClickException("--headers can only be used with --get")
 
@@ -680,6 +670,16 @@ def serve(
         raise click.ClickException("--token can only be used with --get")
 
     if get:
+        # Run async soundness checks before startup hooks, since invoke_startup
+        # now populates internal tables which requires querying each database
+        run_sync(lambda: check_databases(ds))
+
+        # Run the "startup" plugin hooks
+        try:
+            run_sync(ds.invoke_startup)
+        except StartupError as e:
+            raise click.ClickException(e.args[0])
+
         client = TestClient(ds)
         request_headers = {}
         if token:
@@ -704,34 +704,52 @@ def serve(
         sys.exit(exit_code)
         return
 
-    # Start the server
-    url = None
-    if root:
-        ds.root_enabled = True
-        url = "http://{}:{}{}?token={}".format(
-            host, port, ds.urls.path("-/auth-token"), ds._root_token
-        )
-        click.echo(url)
-    if open_browser:
-        if url is None:
-            # Figure out most convenient URL - to table, database or homepage
-            path = run_sync(lambda: initial_path_for_datasette(ds))
-            url = f"http://{host}:{port}{path}"
-        webbrowser.open(url)
-    uvicorn_kwargs = {
-        "host": host,
-        "port": port,
-        "log_level": "info",
-        "lifespan": "on",
-        "workers": 1,
-    }
-    if uds:
-        uvicorn_kwargs["uds"] = uds
-    if ssl_keyfile:
-        uvicorn_kwargs["ssl_keyfile"] = ssl_keyfile
-    if ssl_certfile:
-        uvicorn_kwargs["ssl_certfile"] = ssl_certfile
-    uvicorn.run(ds.app(), **uvicorn_kwargs)
+    # check_databases, invoke_startup() and the uvicorn server all run on a
+    # single event loop, so that anything a plugin's "startup" hook schedules
+    # on the loop (asyncio.create_task, Lock/Queue/Event objects, ...) is
+    # still alive when the server starts handling requests.
+    async def _serve_async():
+        # Run async soundness checks before startup hooks, since invoke_startup
+        # now populates internal tables which requires querying each database
+        await check_databases(ds)
+
+        # Run the "startup" plugin hooks
+        try:
+            await ds.invoke_startup()
+        except StartupError as e:
+            raise click.ClickException(e.args[0])
+
+        # Start the server
+        url = None
+        if root:
+            ds.root_enabled = True
+            url = "http://{}:{}{}?token={}".format(
+                host, port, ds.urls.path("-/auth-token"), ds._root_token
+            )
+            click.echo(url)
+        if open_browser:
+            if url is None:
+                # Figure out most convenient URL - to table, database or homepage
+                path = await initial_path_for_datasette(ds)
+                url = f"http://{host}:{port}{path}"
+            webbrowser.open(url)
+        uvicorn_kwargs = {
+            "host": host,
+            "port": port,
+            "log_level": "info",
+            "lifespan": "on",
+            "workers": 1,
+        }
+        if uds:
+            uvicorn_kwargs["uds"] = uds
+        if ssl_keyfile:
+            uvicorn_kwargs["ssl_keyfile"] = ssl_keyfile
+        if ssl_certfile:
+            uvicorn_kwargs["ssl_certfile"] = ssl_certfile
+        server = uvicorn.Server(uvicorn.Config(ds.app(), **uvicorn_kwargs))
+        await server.serve()
+
+    asyncio.run(_serve_async())
 
 
 @cli.command()
diff --git a/pyproject.toml b/pyproject.toml
index cf5db905..e658955f 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -30,7 +30,7 @@ dependencies = [
     "hupper>=1.9",
     "httpx>=0.20,<1.0",
     "pluggy>=1.0",
-    "uvicorn>=0.11",
+    "uvicorn>=0.29",
     "aiofiles>=0.4",
     "PyYAML>=5.3",
     "mergedeep>=1.1.1",
diff --git a/tests/test_cli_serve_server.py b/tests/test_cli_serve_server.py
index b7604bb8..19fc9198 100644
--- a/tests/test_cli_serve_server.py
+++ b/tests/test_cli_serve_server.py
@@ -1,4 +1,8 @@
 import socket
+import subprocess
+import sys
+import tempfile
+import time
 
 import httpx
 import pytest
@@ -28,3 +32,180 @@ def test_serve_unix_domain_socket(ds_unix_domain_socket_server):
         "path": "/_memory",
         "tables": [],
     }.items() <= response.json().items()
+
+
+def _find_free_port():
+    with socket.socket() as sock:
+        sock.bind(("127.0.0.1", 0))
+        return sock.getsockname()[1]
+
+
+# Shaped after datasette-litestream's (sync) startup hook, which schedules a
+# background task with asyncio.get_running_loop().create_task(...):
+# https://github.com/simonw/datasette-litestream/blob/main/datasette_litestream/__init__.py
+# That only has a chance to actually run if invoke_startup() executes on the
+# same event loop that goes on to serve requests - if it runs on a throwaway
+# loop that gets closed straight after (as on unmodified main), the task is
+# scheduled but never gets a turn before the loop is torn down.
+MARKER_TASK_PLUGIN = '''
+import asyncio
+from datasette import hookimpl
+from datasette.utils.asgi import Response
+
+
+@hookimpl
+def startup(datasette):
+    async def _mark():
+        # The await is essential to the regression: a task with no
+        # internal await point can complete during the brief window
+        # between run_until_complete()'s coroutine finishing and the
+        # temporary loop actually stopping, masking the bug this test
+        # guards against. Real background tasks (like
+        # datasette-litestream's credential_refresh_loop) always have an
+        # internal await, and never get to resume once their throwaway
+        # loop is closed.
+        await asyncio.sleep(0.2)
+        datasette._marker_task_ran = True
+
+    asyncio.get_running_loop().create_task(_mark())
+
+
+@hookimpl
+def register_routes():
+    async def marker_status(datasette):
+        return Response.json(
+            {"marker_task_ran": getattr(datasette, "_marker_task_ran", False)}
+        )
+
+    return [(r"^/-/marker-task-ran$", marker_status)]
+'''
+
+
+STARTUP_ERROR_PLUGIN = '''
+from datasette import hookimpl
+from datasette.utils import StartupError
+
+
+@hookimpl
+def startup(datasette):
+    raise StartupError("boom from plugin")
+'''
+
+
+@pytest.mark.serial
+def test_startup_hook_background_task_runs_on_serving_loop(tmp_path):
+    """
+    Litestream-shaped regression test: a startup hook that does
+    asyncio.get_running_loop().create_task(...) must have that task
+    actually execute before/while the server is handling requests. This
+    only holds if invoke_startup() and uvicorn.Server.serve() share one
+    event loop. This test fails against unmodified main, where
+    invoke_startup() runs on a throwaway loop that is closed before
+    uvicorn opens its own loop to serve.
+    """
+    plugins_dir = tmp_path / "plugins"
+    plugins_dir.mkdir()
+    (plugins_dir / "marker_task_plugin.py").write_text(MARKER_TASK_PLUGIN, "utf-8")
+
+    port = _find_free_port()
+    ds_proc = subprocess.Popen(
+        [
+            sys.executable,
+            "-m",
+            "datasette",
+            "--memory",
+            "--plugins-dir",
+            str(plugins_dir),
+            "-p",
+            str(port),
+        ],
+        stdout=subprocess.PIPE,
+        stderr=subprocess.STDOUT,
+        cwd=tempfile.gettempdir(),
+    )
+    try:
+        url = f"http://localhost:{port}/-/marker-task-ran"
+        deadline = time.time() + 15.0
+        marker_task_ran = False
+        while time.time() < deadline:
+            if ds_proc.poll() is not None:
+                raise AssertionError(
+                    "datasette serve exited early\n"
+                    + ds_proc.stdout.read().decode("utf-8")
+                )
+            try:
+                response = httpx.get(url, timeout=1.0)
+            except httpx.TransportError:
+                time.sleep(0.1)
+                continue
+            if response.status_code == 200 and response.json().get(
+                "marker_task_ran"
+            ):
+                marker_task_ran = True
+                break
+            time.sleep(0.1)
+        assert marker_task_ran, (
+            "The startup hook's asyncio.create_task(...) never ran - "
+            "invoke_startup() and the server are not sharing an event loop"
+        )
+    finally:
+        ds_proc.terminate()
+        try:
+            ds_proc.wait(timeout=5)
+        except subprocess.TimeoutExpired:
+            ds_proc.kill()
+            ds_proc.wait()
+
+
+@pytest.mark.serial
+def test_startup_error_fails_fast_before_port_binds(tmp_path):
+    """
+    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 failure must happen before uvicorn.Server binds the socket.
+    """
+    plugins_dir = tmp_path / "plugins"
+    plugins_dir.mkdir()
+    (plugins_dir / "startup_error_plugin.py").write_text(
+        STARTUP_ERROR_PLUGIN, "utf-8"
+    )
+
+    port = _find_free_port()
+    ds_proc = subprocess.Popen(
+        [
+            sys.executable,
+            "-m",
+            "datasette",
+            "--memory",
+            "--plugins-dir",
+            str(plugins_dir),
+            "-p",
+            str(port),
+        ],
+        stdout=subprocess.PIPE,
+        stderr=subprocess.STDOUT,
+        cwd=tempfile.gettempdir(),
+    )
+    try:
+        deadline = time.time() + 15.0
+        # While the process is still alive (it should crash almost
+        # immediately) repeatedly confirm nothing is listening on the port
+        while ds_proc.poll() is None and time.time() < deadline:
+            with pytest.raises(OSError):
+                with socket.create_connection(("127.0.0.1", port), timeout=0.2):
+                    pass
+            time.sleep(0.05)
+
+        stdout, _ = ds_proc.communicate(timeout=5)
+        output = stdout.decode("utf-8")
+        assert ds_proc.returncode not in (0, None), output
+        assert "boom from plugin" in output, output
+
+        # And confirm it never accepted a connection even now it has exited
+        with pytest.raises(OSError):
+            with socket.create_connection(("127.0.0.1", port), timeout=0.2):
+                pass
+    finally:
+        if ds_proc.poll() is None:
+            ds_proc.kill()
+            ds_proc.wait()

From ba3756eb546b3c0016f6d1c38192207d80fd872e Mon Sep 17 00:00:00 2001
From: Alex Garcia <alexsebastian.garcia@gmail.com>
Date: Mon, 3 Aug 2026 16:52:48 -0700
Subject: [PATCH 2/8] Move the serve-subprocess test plumbing into a conftest
 fixture

The two new tests span a datasette serve subprocess by hand: each found a
free port with a copy of test_playwright.py's find_free_port, built its own
subprocess.Popen call, polled with its own 15 second deadline loop, and
tore the process down in its own finally block. That is the third and
fourth hand-rolled copy of plumbing conftest.py already owns for
ds_localhost_http_server and ds_unix_domain_socket_server.

Move find_free_port into conftest.py and add a serve_with_plugins factory
fixture that writes plugin sources to a temporary --plugins-dir, takes a
free port, waits for the server to answer, and terminates every process it
started when the test ends. wait_until_responds() grows an optional
process argument so a server that dies during startup fails immediately
with its captured output instead of waiting out the timeout, and now
catches httpx.TransportError rather than only httpx.ConnectError - a
superclass, so existing callers are unaffected.

Two fixes beyond the deduplication:

The marker test polls until its flag flips, which meant it would also have
passed if the startup hook were re-run on the serving loop by the
first-request fallback - the exact bug it exists to catch. That cannot
happen while invoke_startup() is idempotent, but nothing said so. The
plugin now counts startup calls and the test asserts it ran exactly once,
so removing that guard fails the test loudly instead of quietly turning it
into a no-op.

test_startup_error_fails_fast_before_port_binds passes on unmodified main,
where startup already ran ahead of uvicorn.run(), so it is a
characterization test rather than a regression test for this commit; its
docstring now says so. Its loop re-checking that nothing was listening ran
about one iteration before the process exited, and could not distinguish a
pre-bind failure from a port nothing ever touched, so it is replaced by a
single check with a comment about what it does and does not prove.

Verified the red side is preserved: the marker test still fails on
unmodified main, now in 3.7s rather than 15.2s.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
---
 tests/conftest.py              |  80 +++++++++++++++-
 tests/test_cli_serve_server.py | 166 +++++++++++----------------------
 2 files changed, 135 insertions(+), 111 deletions(-)

diff --git a/tests/conftest.py b/tests/conftest.py
index a2e6aba2..6b0608ee 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -2,6 +2,7 @@ import importlib.metadata
 import os
 import pathlib
 import re
+import socket
 import subprocess
 import sys
 import tempfile
@@ -32,17 +33,31 @@ UNDOCUMENTED_PERMISSIONS = {
 }
 
 
-def wait_until_responds(url, timeout=5.0, client=httpx, **kwargs):
+def wait_until_responds(url, timeout=5.0, client=httpx, process=None, **kwargs):
     start = time.time()
     while time.time() - start < timeout:
+        # If the server died there is no point waiting out the timeout - fail
+        # now, with its output, instead of after `timeout` seconds of silence
+        if process is not None and process.poll() is not None:
+            raise AssertionError(
+                "Server exited early with returncode {}\n{}".format(
+                    process.returncode, process.stdout.read().decode("utf-8")
+                )
+            )
         try:
             client.get(url, **kwargs)
             return
-        except httpx.ConnectError:
+        except httpx.TransportError:
             time.sleep(0.1)
     raise AssertionError(f"Timed out waiting for {url} to respond")
 
 
+def find_free_port():
+    with socket.socket() as sock:
+        sock.bind(("127.0.0.1", 0))
+        return sock.getsockname()[1]
+
+
 @pytest.fixture
 def bare_ds():
     """
@@ -301,6 +316,67 @@ def ds_unix_domain_socket_server(tmp_path_factory):
             pass
 
 
+@pytest.fixture
+def serve_with_plugins(tmp_path):
+    """Factory fixture for starting ``datasette serve`` in a subprocess with
+    plugins written to a temporary ``--plugins-dir``.
+
+    Unlike ``ds_localhost_http_server`` this is function-scoped and takes a
+    fresh port each time, because each test needs its own plugins. Call it as::
+
+        proc, port = serve_with_plugins({"my_plugin": PLUGIN_SOURCE})
+
+    ``plugins`` maps module name to Python source. Pass
+    ``wait_for_startup=False`` when the server is expected to fail during
+    startup rather than begin serving. Extra CLI arguments are passed through.
+    Every process started is terminated when the test ends.
+    """
+    processes = []
+
+    def start(plugins, *extra_args, wait_for_startup=True):
+        plugins_dir = tmp_path / "plugins"
+        plugins_dir.mkdir(exist_ok=True)
+        for module_name, source in plugins.items():
+            (plugins_dir / "{}.py".format(module_name)).write_text(source, "utf-8")
+        port = find_free_port()
+        proc = subprocess.Popen(
+            [
+                sys.executable,
+                "-m",
+                "datasette",
+                "--memory",
+                "--plugins-dir",
+                str(plugins_dir),
+                "-h",
+                "127.0.0.1",
+                "-p",
+                str(port),
+                *extra_args,
+            ],
+            stdout=subprocess.PIPE,
+            stderr=subprocess.STDOUT,
+            # Avoid FileNotFoundError: [Errno 2] No such file or directory:
+            cwd=tempfile.gettempdir(),
+        )
+        processes.append(proc)
+        if wait_for_startup:
+            wait_until_responds(
+                "http://127.0.0.1:{}/-/versions.json".format(port), process=proc
+            )
+        return proc, port
+
+    yield start
+
+    for proc in processes:
+        if proc.poll() is None:
+            proc.terminate()
+            try:
+                proc.wait(timeout=5)
+            except subprocess.TimeoutExpired:
+                proc.kill()
+                proc.wait()
+
+
 # Import fixtures from fixtures.py to make them available
 from .fixtures import (  # noqa: F401
     TEMP_PLUGIN_SECRET_FILE,
diff --git a/tests/test_cli_serve_server.py b/tests/test_cli_serve_server.py
index 19fc9198..5f74cd9b 100644
--- a/tests/test_cli_serve_server.py
+++ b/tests/test_cli_serve_server.py
@@ -1,7 +1,4 @@
 import socket
-import subprocess
-import sys
-import tempfile
 import time
 
 import httpx
@@ -34,12 +31,6 @@ def test_serve_unix_domain_socket(ds_unix_domain_socket_server):
     }.items() <= response.json().items()
 
 
-def _find_free_port():
-    with socket.socket() as sock:
-        sock.bind(("127.0.0.1", 0))
-        return sock.getsockname()[1]
-
-
 # Shaped after datasette-litestream's (sync) startup hook, which schedules a
 # background task with asyncio.get_running_loop().create_task(...):
 # https://github.com/simonw/datasette-litestream/blob/main/datasette_litestream/__init__.py
@@ -47,7 +38,7 @@ def _find_free_port():
 # same event loop that goes on to serve requests - if it runs on a throwaway
 # loop that gets closed straight after (as on unmodified main), the task is
 # scheduled but never gets a turn before the loop is torn down.
-MARKER_TASK_PLUGIN = '''
+MARKER_TASK_PLUGIN = """
 import asyncio
 from datasette import hookimpl
 from datasette.utils.asgi import Response
@@ -55,6 +46,8 @@ from datasette.utils.asgi import Response
 
 @hookimpl
 def startup(datasette):
+    datasette._startup_calls = getattr(datasette, "_startup_calls", 0) + 1
+
     async def _mark():
         # The await is essential to the regression: a task with no
         # internal await point can complete during the brief window
@@ -74,14 +67,17 @@ def startup(datasette):
 def register_routes():
     async def marker_status(datasette):
         return Response.json(
-            {"marker_task_ran": getattr(datasette, "_marker_task_ran", False)}
+            {
+                "marker_task_ran": getattr(datasette, "_marker_task_ran", False),
+                "startup_calls": getattr(datasette, "_startup_calls", 0),
+            }
         )
 
     return [(r"^/-/marker-task-ran$", marker_status)]
-'''
+"""
 
 
-STARTUP_ERROR_PLUGIN = '''
+STARTUP_ERROR_PLUGIN = """
 from datasette import hookimpl
 from datasette.utils import StartupError
 
@@ -89,11 +85,11 @@ from datasette.utils import StartupError
 @hookimpl
 def startup(datasette):
     raise StartupError("boom from plugin")
-'''
+"""
 
 
 @pytest.mark.serial
-def test_startup_hook_background_task_runs_on_serving_loop(tmp_path):
+def test_startup_hook_background_task_runs_on_serving_loop(serve_with_plugins):
     """
     Litestream-shaped regression test: a startup hook that does
     asyncio.get_running_loop().create_task(...) must have that task
@@ -103,109 +99,61 @@ def test_startup_hook_background_task_runs_on_serving_loop(tmp_path):
     invoke_startup() runs on a throwaway loop that is closed before
     uvicorn opens its own loop to serve.
     """
-    plugins_dir = tmp_path / "plugins"
-    plugins_dir.mkdir()
-    (plugins_dir / "marker_task_plugin.py").write_text(MARKER_TASK_PLUGIN, "utf-8")
-
-    port = _find_free_port()
-    ds_proc = subprocess.Popen(
-        [
-            sys.executable,
-            "-m",
-            "datasette",
-            "--memory",
-            "--plugins-dir",
-            str(plugins_dir),
-            "-p",
-            str(port),
-        ],
-        stdout=subprocess.PIPE,
-        stderr=subprocess.STDOUT,
-        cwd=tempfile.gettempdir(),
+    _, port = serve_with_plugins({"marker_task_plugin": MARKER_TASK_PLUGIN})
+    # The fixture has already waited for the server to answer requests. The
+    # marker task deliberately awaits before setting its flag, so poll for a
+    # moment rather than assuming it landed before the first request arrived.
+    deadline = time.time() + 3.0
+    payload = {}
+    while time.time() < deadline:
+        payload = httpx.get(
+            f"http://127.0.0.1:{port}/-/marker-task-ran", timeout=1.0
+        ).json()
+        if payload["marker_task_ran"]:
+            break
+        time.sleep(0.05)
+    assert payload.get("marker_task_ran"), (
+        "The startup hook's asyncio.create_task(...) never ran - "
+        "invoke_startup() and the server are not sharing an event loop"
     )
-    try:
-        url = f"http://localhost:{port}/-/marker-task-ran"
-        deadline = time.time() + 15.0
-        marker_task_ran = False
-        while time.time() < deadline:
-            if ds_proc.poll() is not None:
-                raise AssertionError(
-                    "datasette serve exited early\n"
-                    + ds_proc.stdout.read().decode("utf-8")
-                )
-            try:
-                response = httpx.get(url, timeout=1.0)
-            except httpx.TransportError:
-                time.sleep(0.1)
-                continue
-            if response.status_code == 200 and response.json().get(
-                "marker_task_ran"
-            ):
-                marker_task_ran = True
-                break
-            time.sleep(0.1)
-        assert marker_task_ran, (
-            "The startup hook's asyncio.create_task(...) never ran - "
-            "invoke_startup() and the server are not sharing an event loop"
+    # Polling above means this test would also pass if the startup hook were
+    # re-run on the serving loop by the first-request fallback - which would
+    # hide exactly the bug being tested. invoke_startup() is idempotent today
+    # so that cannot happen; assert it explicitly so that if the idempotency
+    # guard is ever removed this test fails loudly instead of silently
+    # becoming a no-op.
+    assert payload["startup_calls"] == 1, (
+        "startup hook ran {} times - the marker may have been set by a "
+        "re-run on the serving loop rather than by the original task".format(
+            payload["startup_calls"]
         )
-    finally:
-        ds_proc.terminate()
-        try:
-            ds_proc.wait(timeout=5)
-        except subprocess.TimeoutExpired:
-            ds_proc.kill()
-            ds_proc.wait()
+    )
 
 
 @pytest.mark.serial
-def test_startup_error_fails_fast_before_port_binds(tmp_path):
+def test_startup_error_fails_fast_before_port_binds(serve_with_plugins):
     """
     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 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.
     """
-    plugins_dir = tmp_path / "plugins"
-    plugins_dir.mkdir()
-    (plugins_dir / "startup_error_plugin.py").write_text(
-        STARTUP_ERROR_PLUGIN, "utf-8"
+    proc, port = serve_with_plugins(
+        {"startup_error_plugin": STARTUP_ERROR_PLUGIN}, wait_for_startup=False
     )
+    stdout, _ = proc.communicate(timeout=15)
+    output = stdout.decode("utf-8")
+    assert proc.returncode not in (0, None), output
+    assert "boom from plugin" in output, output
 
-    port = _find_free_port()
-    ds_proc = subprocess.Popen(
-        [
-            sys.executable,
-            "-m",
-            "datasette",
-            "--memory",
-            "--plugins-dir",
-            str(plugins_dir),
-            "-p",
-            str(port),
-        ],
-        stdout=subprocess.PIPE,
-        stderr=subprocess.STDOUT,
-        cwd=tempfile.gettempdir(),
-    )
-    try:
-        deadline = time.time() + 15.0
-        # While the process is still alive (it should crash almost
-        # immediately) repeatedly confirm nothing is listening on the port
-        while ds_proc.poll() is None and time.time() < deadline:
-            with pytest.raises(OSError):
-                with socket.create_connection(("127.0.0.1", port), timeout=0.2):
-                    pass
-            time.sleep(0.05)
-
-        stdout, _ = ds_proc.communicate(timeout=5)
-        output = stdout.decode("utf-8")
-        assert ds_proc.returncode not in (0, None), output
-        assert "boom from plugin" in output, output
-
-        # And confirm it never accepted a connection even now it has exited
-        with pytest.raises(OSError):
-            with socket.create_connection(("127.0.0.1", port), timeout=0.2):
-                pass
-    finally:
-        if ds_proc.poll() is None:
-            ds_proc.kill()
-            ds_proc.wait()
+    # Nothing is listening on the port now the process has exited. This
+    # confirms the socket was not left bound; on its own it cannot prove the
+    # failure preceded the bind, since a port nothing ever touched also
+    # refuses connections.
+    with pytest.raises(OSError):
+        with socket.create_connection(("127.0.0.1", port), timeout=0.2):
+            pass

From d4b6d6cf4e8b30c77a6ec37739f599da5012640c Mon Sep 17 00:00:00 2001
From: Alex Garcia <alexsebastian.garcia@gmail.com>
Date: Mon, 31 Aug 2026 11:39:40 -0700
Subject: [PATCH 3/8] Fix datasette-litestream URL and trim marker-task test
 comments

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012U7coQfVu8nK2R4q2mCULA
---
 tests/test_cli_serve_server.py | 19 +++++--------------
 1 file changed, 5 insertions(+), 14 deletions(-)

diff --git a/tests/test_cli_serve_server.py b/tests/test_cli_serve_server.py
index 5f74cd9b..824e186a 100644
--- a/tests/test_cli_serve_server.py
+++ b/tests/test_cli_serve_server.py
@@ -31,13 +31,9 @@ def test_serve_unix_domain_socket(ds_unix_domain_socket_server):
     }.items() <= response.json().items()
 
 
-# Shaped after datasette-litestream's (sync) startup hook, which schedules a
+# Shaped after datasette-litestream's startup hook, which schedules a
 # background task with asyncio.get_running_loop().create_task(...):
-# https://github.com/simonw/datasette-litestream/blob/main/datasette_litestream/__init__.py
-# That only has a chance to actually run if invoke_startup() executes on the
-# same event loop that goes on to serve requests - if it runs on a throwaway
-# loop that gets closed straight after (as on unmodified main), the task is
-# scheduled but never gets a turn before the loop is torn down.
+# https://github.com/datasette/datasette-litestream
 MARKER_TASK_PLUGIN = """
 import asyncio
 from datasette import hookimpl
@@ -49,14 +45,9 @@ def startup(datasette):
     datasette._startup_calls = getattr(datasette, "_startup_calls", 0) + 1
 
     async def _mark():
-        # The await is essential to the regression: a task with no
-        # internal await point can complete during the brief window
-        # between run_until_complete()'s coroutine finishing and the
-        # temporary loop actually stopping, masking the bug this test
-        # guards against. Real background tasks (like
-        # datasette-litestream's credential_refresh_loop) always have an
-        # internal await, and never get to resume once their throwaway
-        # loop is closed.
+        # Must await before setting the flag: a task with no internal
+        # await point could finish on the throwaway loop before it
+        # closed, masking the regression this test guards against.
         await asyncio.sleep(0.2)
         datasette._marker_task_ran = True
 

From 55b995e5a3777913698e244a1f813d876d4c54ff Mon Sep 17 00:00:00 2001
From: Alex Garcia <alexsebastian.garcia@gmail.com>
Date: Mon, 31 Aug 2026 11:52:27 -0700
Subject: [PATCH 4/8] Explain why serve_with_plugins needs a subprocess and
 plugin files

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012U7coQfVu8nK2R4q2mCULA
---
 tests/conftest.py | 4 ++++
 1 file changed, 4 insertions(+)

diff --git a/tests/conftest.py b/tests/conftest.py
index 6b0608ee..309ba1e4 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -321,6 +321,10 @@ def serve_with_plugins(tmp_path):
     """Factory fixture for starting ``datasette serve`` in a subprocess with
     plugins written to a temporary ``--plugins-dir``.
 
+    For tests that need the real serve path: event-loop wiring, exit codes,
+    signals. The usual in-process ``pm.register`` plugin pattern can't reach
+    a subprocess, so plugin source is written out as importable files instead.
+
     Unlike ``ds_localhost_http_server`` this is function-scoped and takes a
     fresh port each time, because each test needs its own plugins. Call it as::
 

From c748683f2e6b648829e14c6a27656c335b4295ca Mon Sep 17 00:00:00 2001
From: Alex Garcia <alexsebastian.garcia@gmail.com>
Date: Mon, 31 Aug 2026 13:17:44 -0700
Subject: [PATCH 5/8] Apply ruff 0.16 and black fixes

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012U7coQfVu8nK2R4q2mCULA
---
 tests/conftest.py              | 4 ++--
 tests/test_cli_serve_server.py | 8 +++++---
 2 files changed, 7 insertions(+), 5 deletions(-)

diff --git a/tests/conftest.py b/tests/conftest.py
index 309ba1e4..12dce417 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -341,7 +341,7 @@ def serve_with_plugins(tmp_path):
         plugins_dir = tmp_path / "plugins"
         plugins_dir.mkdir(exist_ok=True)
         for module_name, source in plugins.items():
-            (plugins_dir / "{}.py".format(module_name)).write_text(source, "utf-8")
+            (plugins_dir / f"{module_name}.py").write_text(source, "utf-8")
         port = find_free_port()
         proc = subprocess.Popen(
             [
@@ -365,7 +365,7 @@ def serve_with_plugins(tmp_path):
         processes.append(proc)
         if wait_for_startup:
             wait_until_responds(
-                "http://127.0.0.1:{}/-/versions.json".format(port), process=proc
+                f"http://127.0.0.1:{port}/-/versions.json", process=proc
             )
         return proc, port
 
diff --git a/tests/test_cli_serve_server.py b/tests/test_cli_serve_server.py
index 824e186a..2f113ded 100644
--- a/tests/test_cli_serve_server.py
+++ b/tests/test_cli_serve_server.py
@@ -145,6 +145,8 @@ def test_startup_error_fails_fast_before_port_binds(serve_with_plugins):
     # confirms the socket was not left bound; on its own it cannot prove the
     # failure preceded the bind, since a port nothing ever touched also
     # refuses connections.
-    with pytest.raises(OSError):
-        with socket.create_connection(("127.0.0.1", port), timeout=0.2):
-            pass
+    with (
+        pytest.raises(OSError),
+        socket.create_connection(("127.0.0.1", port), timeout=0.2),
+    ):
+        pass

From e78b8a2e6ac69310c06fdacc6ca0a6ab309ffe0b Mon Sep 17 00:00:00 2001
From: Alex Garcia <alexsebastian.garcia@gmail.com>
Date: Tue, 1 Sep 2026 09:32:37 -0700
Subject: [PATCH 6/8] 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
---
 datasette/cli.py               |  91 ++++++++++++++-----------
 pyproject.toml                 |   2 +-
 tests/conftest.py              |  84 ++++++++++++++++++++++-
 tests/test_cli_serve_server.py | 117 +++++++++++++++++++++++++++++++++
 4 files changed, 253 insertions(+), 41 deletions(-)

diff --git a/datasette/cli.py b/datasette/cli.py
index 57db83b6..12024a14 100644
--- a/datasette/cli.py
+++ b/datasette/cli.py
@@ -663,16 +663,6 @@ def serve(
         # Private utility mechanism for writing unit tests
         return ds
 
-    # Run async soundness checks before startup hooks, since invoke_startup
-    # now populates internal tables which requires querying each database
-    run_sync(lambda: check_databases(ds))
-
-    # Run the "startup" plugin hooks
-    try:
-        run_sync(ds.invoke_startup)
-    except StartupError as e:
-        raise click.ClickException(e.args[0])
-
     if headers and not get:
         raise click.ClickException("--headers can only be used with --get")
 
@@ -680,6 +670,14 @@ def serve(
         raise click.ClickException("--token can only be used with --get")
 
     if get:
+        # --get means we don't run Uvicorn at all
+        run_sync(lambda: check_databases(ds))
+
+        try:
+            run_sync(ds.invoke_startup)
+        except StartupError as e:
+            raise click.ClickException(e.args[0])
+
         client = TestClient(ds)
         request_headers = {}
         if token:
@@ -704,34 +702,51 @@ def serve(
         sys.exit(exit_code)
         return
 
-    # Start the server
-    url = None
-    if root:
-        ds.root_enabled = True
-        url = "http://{}:{}{}?token={}".format(
-            host, port, ds.urls.path("-/auth-token"), ds._root_token
-        )
-        click.echo(url)
-    if open_browser:
-        if url is None:
-            # Figure out most convenient URL - to table, database or homepage
-            path = run_sync(lambda: initial_path_for_datasette(ds))
-            url = f"http://{host}:{port}{path}"
-        webbrowser.open(url)
-    uvicorn_kwargs = {
-        "host": host,
-        "port": port,
-        "log_level": "info",
-        "lifespan": "on",
-        "workers": 1,
-    }
-    if uds:
-        uvicorn_kwargs["uds"] = uds
-    if ssl_keyfile:
-        uvicorn_kwargs["ssl_keyfile"] = ssl_keyfile
-    if ssl_certfile:
-        uvicorn_kwargs["ssl_certfile"] = ssl_certfile
-    uvicorn.run(ds.app(), **uvicorn_kwargs)
+    # check_databases, invoke_startup() and the uvicorn server all run on a
+    # single event loop, so that anything a plugin's "startup" hook schedules
+    # on the loop (asyncio.create_task, Lock/Queue/Event objects, ...) is
+    # still alive when the server starts handling requests.
+    async def _serve_async():
+        # Populate internal catalog tables before invoke_startup
+        await check_databases(ds)
+
+        # Run the "startup" plugin hooks
+        try:
+            await ds.invoke_startup()
+        except StartupError as e:
+            raise click.ClickException(e.args[0])
+
+        # Start the server
+        url = None
+        if root:
+            ds.root_enabled = True
+            url = "http://{}:{}{}?token={}".format(
+                host, port, ds.urls.path("-/auth-token"), ds._root_token
+            )
+            click.echo(url)
+        if open_browser:
+            if url is None:
+                # Figure out most convenient URL - to table, database or homepage
+                path = await initial_path_for_datasette(ds)
+                url = f"http://{host}:{port}{path}"
+            webbrowser.open(url)
+        uvicorn_kwargs = {
+            "host": host,
+            "port": port,
+            "log_level": "info",
+            "lifespan": "on",
+            "workers": 1,
+        }
+        if uds:
+            uvicorn_kwargs["uds"] = uds
+        if ssl_keyfile:
+            uvicorn_kwargs["ssl_keyfile"] = ssl_keyfile
+        if ssl_certfile:
+            uvicorn_kwargs["ssl_certfile"] = ssl_certfile
+        server = uvicorn.Server(uvicorn.Config(ds.app(), **uvicorn_kwargs))
+        await server.serve()
+
+    asyncio.run(_serve_async())
 
 
 @cli.command()
diff --git a/pyproject.toml b/pyproject.toml
index cf5db905..e658955f 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -30,7 +30,7 @@ dependencies = [
     "hupper>=1.9",
     "httpx>=0.20,<1.0",
     "pluggy>=1.0",
-    "uvicorn>=0.11",
+    "uvicorn>=0.29",
     "aiofiles>=0.4",
     "PyYAML>=5.3",
     "mergedeep>=1.1.1",
diff --git a/tests/conftest.py b/tests/conftest.py
index a2e6aba2..12dce417 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -2,6 +2,7 @@ import importlib.metadata
 import os
 import pathlib
 import re
+import socket
 import subprocess
 import sys
 import tempfile
@@ -32,17 +33,31 @@ UNDOCUMENTED_PERMISSIONS = {
 }
 
 
-def wait_until_responds(url, timeout=5.0, client=httpx, **kwargs):
+def wait_until_responds(url, timeout=5.0, client=httpx, process=None, **kwargs):
     start = time.time()
     while time.time() - start < timeout:
+        # If the server died there is no point waiting out the timeout - fail
+        # now, with its output, instead of after `timeout` seconds of silence
+        if process is not None and process.poll() is not None:
+            raise AssertionError(
+                "Server exited early with returncode {}\n{}".format(
+                    process.returncode, process.stdout.read().decode("utf-8")
+                )
+            )
         try:
             client.get(url, **kwargs)
             return
-        except httpx.ConnectError:
+        except httpx.TransportError:
             time.sleep(0.1)
     raise AssertionError(f"Timed out waiting for {url} to respond")
 
 
+def find_free_port():
+    with socket.socket() as sock:
+        sock.bind(("127.0.0.1", 0))
+        return sock.getsockname()[1]
+
+
 @pytest.fixture
 def bare_ds():
     """
@@ -301,6 +316,71 @@ def ds_unix_domain_socket_server(tmp_path_factory):
             pass
 
 
+@pytest.fixture
+def serve_with_plugins(tmp_path):
+    """Factory fixture for starting ``datasette serve`` in a subprocess with
+    plugins written to a temporary ``--plugins-dir``.
+
+    For tests that need the real serve path: event-loop wiring, exit codes,
+    signals. The usual in-process ``pm.register`` plugin pattern can't reach
+    a subprocess, so plugin source is written out as importable files instead.
+
+    Unlike ``ds_localhost_http_server`` this is function-scoped and takes a
+    fresh port each time, because each test needs its own plugins. Call it as::
+
+        proc, port = serve_with_plugins({"my_plugin": PLUGIN_SOURCE})
+
+    ``plugins`` maps module name to Python source. Pass
+    ``wait_for_startup=False`` when the server is expected to fail during
+    startup rather than begin serving. Extra CLI arguments are passed through.
+    Every process started is terminated when the test ends.
+    """
+    processes = []
+
+    def start(plugins, *extra_args, wait_for_startup=True):
+        plugins_dir = tmp_path / "plugins"
+        plugins_dir.mkdir(exist_ok=True)
+        for module_name, source in plugins.items():
+            (plugins_dir / f"{module_name}.py").write_text(source, "utf-8")
+        port = find_free_port()
+        proc = subprocess.Popen(
+            [
+                sys.executable,
+                "-m",
+                "datasette",
+                "--memory",
+                "--plugins-dir",
+                str(plugins_dir),
+                "-h",
+                "127.0.0.1",
+                "-p",
+                str(port),
+                *extra_args,
+            ],
+            stdout=subprocess.PIPE,
+            stderr=subprocess.STDOUT,
+            # Avoid FileNotFoundError: [Errno 2] No such file or directory:
+            cwd=tempfile.gettempdir(),
+        )
+        processes.append(proc)
+        if wait_for_startup:
+            wait_until_responds(
+                f"http://127.0.0.1:{port}/-/versions.json", process=proc
+            )
+        return proc, port
+
+    yield start
+
+    for proc in processes:
+        if proc.poll() is None:
+            proc.terminate()
+            try:
+                proc.wait(timeout=5)
+            except subprocess.TimeoutExpired:
+                proc.kill()
+                proc.wait()
+
+
 # Import fixtures from fixtures.py to make them available
 from .fixtures import (  # noqa: F401
     TEMP_PLUGIN_SECRET_FILE,
diff --git a/tests/test_cli_serve_server.py b/tests/test_cli_serve_server.py
index b7604bb8..b76180fd 100644
--- a/tests/test_cli_serve_server.py
+++ b/tests/test_cli_serve_server.py
@@ -1,4 +1,5 @@
 import socket
+import time
 
 import httpx
 import pytest
@@ -28,3 +29,119 @@ def test_serve_unix_domain_socket(ds_unix_domain_socket_server):
         "path": "/_memory",
         "tables": [],
     }.items() <= response.json().items()
+
+
+# Shaped after datasette-litestream's startup hook, which schedules a
+# background task with asyncio.get_running_loop().create_task(...):
+# https://github.com/datasette/datasette-litestream
+MARKER_TASK_PLUGIN = """
+import asyncio
+from datasette import hookimpl
+from datasette.utils.asgi import Response
+
+
+@hookimpl
+def startup(datasette):
+    datasette._startup_calls = getattr(datasette, "_startup_calls", 0) + 1
+
+    async def _mark():
+        # Must await before setting the flag: a task with no internal
+        # await point could finish on the throwaway loop before it
+        # closed, masking the regression this test guards against.
+        await asyncio.sleep(0.2)
+        datasette._marker_task_ran = True
+
+    asyncio.get_running_loop().create_task(_mark())
+
+
+@hookimpl
+def register_routes():
+    async def marker_status(datasette):
+        return Response.json(
+            {
+                "marker_task_ran": getattr(datasette, "_marker_task_ran", False),
+                "startup_calls": getattr(datasette, "_startup_calls", 0),
+            }
+        )
+
+    return [(r"^/-/marker-task-ran$", marker_status)]
+"""
+
+
+STARTUP_ERROR_PLUGIN = """
+from datasette import hookimpl
+from datasette.utils import StartupError
+
+
+@hookimpl
+def startup(datasette):
+    raise StartupError("boom from plugin")
+"""
+
+
+@pytest.mark.serial
+def test_startup_hook_background_task_runs_on_serving_loop(serve_with_plugins):
+    """
+    Litestream-shaped regression test: a startup hook that does
+    asyncio.get_running_loop().create_task(...) must have that task
+    actually execute before/while the server is handling requests. This
+    only holds if invoke_startup() and uvicorn.Server.serve() share one
+    event loop. This test fails against unmodified main, where
+    invoke_startup() runs on a throwaway loop that is closed before
+    uvicorn opens its own loop to serve.
+    """
+    _, port = serve_with_plugins({"marker_task_plugin": MARKER_TASK_PLUGIN})
+    # The fixture has already waited for the server to answer requests. The
+    # marker task deliberately awaits before setting its flag, so poll for a
+    # moment rather than assuming it landed before the first request arrived.
+    deadline = time.time() + 3.0
+    payload = {}
+    while time.time() < deadline:
+        payload = httpx.get(
+            f"http://127.0.0.1:{port}/-/marker-task-ran", timeout=1.0
+        ).json()
+        if payload["marker_task_ran"]:
+            break
+        time.sleep(0.05)
+    assert payload.get("marker_task_ran"), (
+        "The startup hook's asyncio.create_task(...) never ran - "
+        "invoke_startup() and the server are not sharing an event loop"
+    )
+    # Polling above means this test would also pass if the startup hook were
+    # re-run on the serving loop by the first-request fallback - which would
+    # hide exactly the bug being tested. invoke_startup() is idempotent today
+    # so that cannot happen; assert it explicitly so that if the idempotency
+    # guard is ever removed this test fails loudly instead of silently
+    # becoming a no-op.
+    assert payload["startup_calls"] == 1, (
+        "startup hook ran {} times - the marker may have been set by a "
+        "re-run on the serving loop rather than by the original task".format(
+            payload["startup_calls"]
+        )
+    )
+
+
+@pytest.mark.serial
+def test_startup_error_fails_fast_before_port_binds(serve_with_plugins):
+    """
+    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 failure must happen before uvicorn.Server binds the socket.
+    """
+    proc, port = serve_with_plugins(
+        {"startup_error_plugin": STARTUP_ERROR_PLUGIN}, wait_for_startup=False
+    )
+    stdout, _ = proc.communicate(timeout=15)
+    output = stdout.decode("utf-8")
+    assert proc.returncode not in (0, None), output
+    assert "boom from plugin" in output, output
+
+    # Nothing is listening on the port now the process has exited. This
+    # confirms the socket was not left bound; on its own it cannot prove the
+    # failure preceded the bind, since a port nothing ever touched also
+    # refuses connections.
+    with (
+        pytest.raises(OSError),
+        socket.create_connection(("127.0.0.1", port), timeout=0.2),
+    ):
+        pass

From 3e018bb1b571cef87c67ae718a5472f23d6b3c6f Mon Sep 17 00:00:00 2001
From: Alex Garcia <alexsebastian.garcia@gmail.com>
Date: Tue, 1 Sep 2026 09:39:25 -0700
Subject: [PATCH 7/8] 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
---
 datasette/app.py        |  47 ++++++--
 datasette/cli.py        |   7 +-
 datasette/utils/asgi.py |  42 +++++--
 tests/test_lifespan.py  | 259 ++++++++++++++++++++++++++++++++++++++++
 4 files changed, 337 insertions(+), 18 deletions(-)
 create mode 100644 tests/test_lifespan.py

diff --git a/datasette/app.py b/datasette/app.py
index c82ea075..42be7425 100644
--- a/datasette/app.py
+++ b/datasette/app.py
@@ -453,8 +453,10 @@ class Datasette:
         self.databases = collections.OrderedDict()
         self.actions = {}  # .invoke_startup() will populate this
         self._column_types = {}  # .invoke_startup() will populate this
+        self._setup_db_done = False
         try:
             self._refresh_schemas_lock = asyncio.Lock()
+            self._startup_lock = asyncio.Lock()
         except RuntimeError as rex:
             # Workaround for intermittent test failure, see:
             # https://github.com/simonw/datasette/issues/1802
@@ -462,6 +464,7 @@ class Datasette:
                 loop = asyncio.new_event_loop()
                 asyncio.set_event_loop(loop)
                 self._refresh_schemas_lock = asyncio.Lock()
+                self._startup_lock = asyncio.Lock()
             else:
                 raise
         self.crossdb = crossdb
@@ -2803,24 +2806,52 @@ class Datasette:
             raise RowNotFound(db.name, table_name, pk_values)
         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):
         """Returns an ASGI app function that serves the whole of Datasette"""
         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():
             self.close()
 
         asgi = CrossOriginProtectionMiddleware(DatasetteRouter(self, routes), self)
         if self.setting("trace_debug"):
             asgi = AsgiTracer(asgi)
-        asgi = AsgiLifespan(asgi, on_shutdown=[_close_on_shutdown])
-        asgi = AsgiRunOnFirstRequest(asgi, on_startup=[setup_db, self.invoke_startup])
+        asgi = AsgiLifespan(
+            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):
             asgi = wrapper(asgi)
         return asgi
diff --git a/datasette/cli.py b/datasette/cli.py
index 12024a14..2694c1f6 100644
--- a/datasette/cli.py
+++ b/datasette/cli.py
@@ -710,9 +710,12 @@ def serve(
         # Populate internal catalog tables before invoke_startup
         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:
-            await ds.invoke_startup()
+            await ds._startup_sequence()
         except StartupError as e:
             raise click.ClickException(e.args[0])
 
diff --git a/datasette/utils/asgi.py b/datasette/utils/asgi.py
index 812194fd..2614ad02 100644
--- a/datasette/utils/asgi.py
+++ b/datasette/utils/asgi.py
@@ -1,3 +1,4 @@
+import asyncio
 import json
 import re
 from http.cookies import Morsel, SimpleCookie
@@ -300,12 +301,24 @@ class AsgiLifespan:
             while True:
                 message = await receive()
                 if message["type"] == "lifespan.startup":
-                    for fn in self.on_startup:
-                        await fn()
+                    try:
+                        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"})
                 elif message["type"] == "lifespan.shutdown":
-                    for fn in self.on_shutdown:
-                        await fn()
+                    try:
+                        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"})
                     return
         else:
@@ -624,10 +637,23 @@ class AsgiRunOnFirstRequest:
         self.asgi = asgi
         self.on_startup = on_startup
         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):
-        if not self._started:
-            self._started = True
-            for hook in self.on_startup:
-                await hook()
+        # Leave "lifespan" scope events alone - this shim only exists as a
+        # fallback for hosts that never send them. It wraps AsgiLifespan, so
+        # if it ran on_startup here too, a startup exception would escape
+        # 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)
diff --git a/tests/test_lifespan.py b/tests/test_lifespan.py
new file mode 100644
index 00000000..3655285e
--- /dev/null
+++ b/tests/test_lifespan.py
@@ -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

From bdc973174096cae350ddaa733a10ed8b3ffd970b Mon Sep 17 00:00:00 2001
From: Simon Willison <swillison@gmail.com>
Date: Tue, 1 Sep 2026 13:37:15 -0700
Subject: [PATCH 8/8] check-latest: true, add 3.15 to test matrix, to test RCs
 (#2895)

See https://simonwillison.net/2026/Sep/1/python-315-rc-2/
---
 .github/workflows/test.yml | 5 +++--
 1 file changed, 3 insertions(+), 2 deletions(-)

diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml
index 751eedfd..2a8c0ae4 100644
--- a/.github/workflows/test.yml
+++ b/.github/workflows/test.yml
@@ -11,16 +11,17 @@ jobs:
     strategy:
       fail-fast: false
       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:
     - uses: actions/checkout@v7
     - name: Set up Python ${{ matrix.python-version }}
-      uses: actions/setup-python@v6
+      uses: actions/setup-python@v7
       with:
         python-version: ${{ matrix.python-version }}
         allow-prereleases: true
         cache: pip
         cache-dependency-path: pyproject.toml
+        check-latest: true
     - name: Build extension for --load-extension test
       run: |-
         (cd tests && gcc ext.c -fPIC -shared -o ext.so)