Apply ruff 0.16 and black fixes

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012U7coQfVu8nK2R4q2mCULA

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

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,55 @@ 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 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._startup_sequence()
+        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()

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)

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",

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,

tests/test_cli_serve_server.py

@@ -1,4 +1,5 @@
 import socket
+import time
 
 import httpx
 import pytest
@@ -28,3 +29,124 @@ 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.
+
+    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(
+        {"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

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