diff --git a/datasette/app.py b/datasette/app.py
index 8cee9b74..61d3cd37 100644
--- a/datasette/app.py
+++ b/datasette/app.py
@@ -52,8 +52,10 @@ from .resources import DatabaseResource, TableResource
 from .telemetry import (
     TelemetryMiddleware,
     clamp_http_method,
+    register_datasette,
     request_span,
     tracer,
+    unregister_datasette,
 )
 from .telemetry_registry import HTTP_ROUTE, STARTUP
 from .tokens import TokenInvalid
@@ -639,6 +641,10 @@ class Datasette:
         self.root_enabled = False
         self.default_deny = default_deny
         self.client = DatasetteClient(self)
+        # Last, so that the observable-gauge callbacks - which may fire on the
+        # SDK's collection thread the instant this returns - never see a
+        # half-built instance.
+        register_datasette(self)
 
     async def apply_metadata_json(self):
         # Apply any metadata entries from metadata.json to the internal tables
@@ -978,6 +984,10 @@ class Datasette:
         if self._closed:
             return
         self._closed = True
+        # Stop reporting gauges before tearing anything down, so a collection
+        # cycle landing mid-close cannot observe a half-closed instance. The
+        # WeakSet would drop it eventually anyway; this makes it immediate.
+        unregister_datasette(self)
         first_exception = None
         dbs = list(self.databases.values()) + [self._internal_database]
         for db in dbs:
diff --git a/datasette/database.py b/datasette/database.py
index efb8c56e..cfa38d11 100644
--- a/datasette/database.py
+++ b/datasette/database.py
@@ -17,7 +17,14 @@ from opentelemetry import context as otel_context_api
 from opentelemetry.trace import Link, Status, StatusCode, get_current_span
 
 from .inspect import inspect_hash
-from .telemetry import sql_attribute, sql_operation_name, tracer
+from .telemetry import (
+    record_operation_duration,
+    record_query_interrupted,
+    record_write_queue_wait,
+    sql_attribute,
+    sql_operation_name,
+    tracer,
+)
 from .telemetry_registry import (
     DB_COLLECTION_NAME,
     DB_NAMESPACE,
@@ -292,9 +299,10 @@ class Database:
                 span.set_attribute(DB_OPERATION_NAME, operation_name)
             if params:
                 span.set_attribute(PARAM_COUNT, len(params))
-            results = await self.execute_write_fn(
-                _inner, block=block, request=request, transaction=transaction
-            )
+            with record_operation_duration(self.name, "write"):
+                results = await self.execute_write_fn(
+                    _inner, block=block, request=request, transaction=transaction
+                )
         return results
 
     async def execute_write_script(self, sql, block=True, request=None):
@@ -312,9 +320,10 @@ class Database:
             span.set_attribute(DB_NAMESPACE, self.name)
             span.set_attribute(DB_QUERY_TEXT, sql_attribute(sql))
             span.set_attribute(EXECUTESCRIPT, True)
-            results = await self.execute_write_fn(
-                _inner, block=block, transaction=False, request=request
-            )
+            with record_operation_duration(self.name, "write"):
+                results = await self.execute_write_fn(
+                    _inner, block=block, transaction=False, request=request
+                )
         return results
 
     async def execute_write_many(self, sql, params_seq, block=True, request=None):
@@ -341,9 +350,10 @@ class Database:
             operation_name = sql_operation_name(sql)
             if operation_name:
                 span.set_attribute(DB_OPERATION_NAME, operation_name)
-            results, count = await self.execute_write_fn(
-                _inner, block=block, request=request
-            )
+            with record_operation_duration(self.name, "write"):
+                results, count = await self.execute_write_fn(
+                    _inner, block=block, request=request
+                )
             # count is the number of parameter *sets* consumed by
             # executemany(), not a row count - executemany returns no rows.
             span.set_attribute(PARAM_SETS, count)
@@ -596,11 +606,15 @@ class Database:
                 # this span's duration is the time the task actually spent
                 # waiting in the queue (enqueue -> dequeue), not the near-
                 # zero time spent constructing/ending the span object here.
+                dequeued_at_ns = time.time_ns()
                 tracer.start_span(
                     DB_WRITE_QUEUE_WAIT,
                     start_time=task.enqueued_at_ns,
                     **write_span_kwargs,
-                ).end(end_time=time.time_ns())
+                ).end(end_time=dequeued_at_ns)
+                record_write_queue_wait(
+                    self.name, dequeued_at_ns - task.enqueued_at_ns
+                )
                 if conn_exception is not None:
                     # fn never runs in this branch, so there is nothing to
                     # wrap in a db.write.execute span.
@@ -822,7 +836,8 @@ class Database:
             if params:
                 span.set_attribute(PARAM_COUNT, len(params))
             try:
-                results = await self.execute_fn(sql_operation_in_thread)
+                with record_operation_duration(self.name, "read"):
+                    results = await self.execute_fn(sql_operation_in_thread)
             except QueryInterrupted as e:
                 # datasette.interrupted is set either way - it is the
                 # signal worth having. Only the ERROR status is
@@ -831,6 +846,14 @@ class Database:
                 if not timeout_expected:
                     span.set_status(Status(StatusCode.ERROR, str(e)))
                     span.record_exception(e)
+                    # A counter rather than only a span, because this is the
+                    # one thing an operator wants a rate and an alert on, and
+                    # spans under a 1% sampler cannot provide either. An
+                    # expected timeout - a caller that opted into a shorter
+                    # budget, like facet suggestion - is not counted, for the
+                    # same reason it is not a span error: it fires routinely
+                    # by design and would drown the signal this exists for.
+                    record_query_interrupted(self.name)
                 raise
             except Exception as e:
                 # log_sql_errors=False means the caller is probing and
diff --git a/datasette/telemetry.py b/datasette/telemetry.py
index 311d661f..b2d7a44a 100644
--- a/datasette/telemetry.py
+++ b/datasette/telemetry.py
@@ -2,9 +2,9 @@
 OpenTelemetry integration for Datasette core.
 
 Core depends on `opentelemetry-api` only. It never creates a
-`TracerProvider`, never configures an exporter, and never touches
-sampling - that is the responsibility of whoever is running Datasette
-(an `opentelemetry-instrument` agent, a future plugin, or a test
+`TracerProvider` or a `MeterProvider`, never configures an exporter, and
+never touches sampling - that is the responsibility of whoever is running
+Datasette (an `opentelemetry-instrument` agent, a future plugin, or a test
 harness).
 
 With no provider installed every span produced here is a
@@ -12,17 +12,26 @@ With no provider installed every span produced here is a
 but it is below what an end-to-end page benchmark can resolve: measured
 across 15 runs of a 5,000-row table page, the median moved 9.80ms to
 9.98ms while run-to-run spread was 1.4ms. Installing an SDK provider is
-what costs something measurable.
+what costs something measurable. Every metric instrument is likewise a
+no-op without a provider, and the observable-gauge callbacks are never
+invoked at all.
 """
 
 import re
+import threading
+import time
+import weakref
+from contextlib import contextmanager
 
+from opentelemetry import metrics as otel_metrics
 from opentelemetry import trace as otel_trace
 from opentelemetry.propagate import extract
 from opentelemetry.propagators.textmap import Getter
 from opentelemetry.trace import SpanKind, Status, StatusCode
 
 from .telemetry_registry import (
+    DB_NAMESPACE,
+    DB_SYSTEM,
     ERROR_TYPE,
     HTTP_REQUEST_METHOD,
     HTTP_RESPONSE_STATUS_CODE,
@@ -55,6 +64,7 @@ from .version import __version__
 SCHEMA_URL = "https://opentelemetry.io/schemas/1.29.0"
 
 tracer = otel_trace.get_tracer("datasette", __version__, schema_url=SCHEMA_URL)
+meter = otel_metrics.get_meter("datasette", __version__, schema_url=SCHEMA_URL)
 
 MAX_SQL_LENGTH = 2048
 
@@ -345,3 +355,249 @@ class TelemetryMiddleware:
                     if status >= 500 and not escaped:
                         span.set_status(Status(StatusCode.ERROR))
                         span.set_attribute(ERROR_TYPE, str(status))
+
+
+# --- Metrics --------------------------------------------------------------
+#
+# Spans answer "what happened during this request". They cannot answer "am I
+# saturating my 3 SQL threads right now", because that is a gauge: a level
+# sampled at collection time, not an event with a duration. It is also the
+# single most useful operational question about a Datasette deployment, since
+# num_sql_threads defaults to 3 and every read query in the process competes
+# for those threads.
+#
+# Two shapes are used here:
+#
+#   Observable gauges - a callback the SDK invokes on its own collection
+#   cycle. Nothing is computed unless something is collecting, so the default
+#   no-provider install pays literally nothing for them.
+#
+#   Synchronous histograms/counters - recorded inline on the query path. These
+#   survive trace sampling, which spans do not: an operator sampling 1% of
+#   traces still gets 100% of the latency distribution and the interrupted
+#   count.
+#
+# Note a real difference from tracing: `_ProxyMeter` and its instruments
+# forward to a provider installed *after* they were created, whereas
+# `ProxyTracer` permanently caches the concrete tracer it first resolves. So
+# module-level instruments here are safe, and tests do not need a provider
+# installed before this module is imported.
+
+
+def _duration_attributes(database_name, operation):
+    return {
+        DB_SYSTEM: "sqlite",
+        DB_NAMESPACE: database_name,
+        "datasette.operation": operation,
+    }
+
+
+sql_operation_duration = meter.create_histogram(
+    "db.client.operation.duration",
+    unit="s",
+    description="Duration of a SQL operation issued by Datasette",
+)
+
+write_queue_wait = meter.create_histogram(
+    "datasette.write.queue_wait",
+    unit="s",
+    description=(
+        "Time a write spent queued behind the single write thread for its database"
+    ),
+)
+
+queries_interrupted = meter.create_counter(
+    "datasette.sql.queries.interrupted",
+    unit="{query}",
+    description=(
+        "Queries cancelled for exceeding sql_time_limit_ms. Not derivable from "
+        "spans under sampling, and the signal that a time limit is too tight"
+    ),
+)
+
+
+@contextmanager
+def record_operation_duration(database_name, operation):
+    """
+    Record `db.client.operation.duration` for one SQL operation.
+
+    `error.type` is set from the exception class on failure, per semconv, so a
+    latency distribution can be split by success and failure. For a
+    `block=False` write this measures the enqueue, not the write - the same
+    caveat that applies to the surrounding span.
+    """
+    attributes = _duration_attributes(database_name, operation)
+    started = time.perf_counter()
+    try:
+        yield
+    except BaseException as exception:
+        attributes[ERROR_TYPE] = type(exception).__qualname__
+        raise
+    finally:
+        sql_operation_duration.record(time.perf_counter() - started, attributes)
+
+
+def record_write_queue_wait(database_name, waited_ns):
+    write_queue_wait.record(waited_ns / 1e9, {DB_NAMESPACE: database_name})
+
+
+def record_query_interrupted(database_name):
+    queries_interrupted.add(1, {DB_NAMESPACE: database_name})
+
+
+# Live Datasette instances, weakly held so that instrumenting an instance
+# never keeps it alive. Guarded by a lock because the gauge callbacks run on
+# the SDK's collection thread while the event loop may be building or closing
+# a Datasette.
+#
+# Known limitation: the pool gauges below carry no attribute identifying which
+# Datasette produced them, so if a single process runs more than one instance
+# their observations collide and last-one-wins. Production runs one instance
+# per process; adding an instance id to make the test suite's hundreds of
+# instances distinguishable would mean unbounded attribute cardinality in
+# exchange for fixing a case that does not occur in production.
+_live_datasettes = weakref.WeakSet()
+_live_datasettes_lock = threading.Lock()
+
+
+def register_datasette(ds):
+    "Start reporting pool/queue gauges for this Datasette instance."
+    with _live_datasettes_lock:
+        _live_datasettes.add(ds)
+
+
+def unregister_datasette(ds):
+    "Stop reporting gauges for an instance that has been closed."
+    with _live_datasettes_lock:
+        _live_datasettes.discard(ds)
+
+
+def _live_instances():
+    with _live_datasettes_lock:
+        return list(_live_datasettes)
+
+
+def _databases_of(ds):
+    """
+    Every Database attached to an instance, including the internal database.
+
+    The internal database is deliberately included: permission checks run SQL
+    against it on essentially every request, so its queue depth and connection
+    count are as operationally interesting as any user database's.
+    """
+    databases = list(ds.databases.values())
+    internal = getattr(ds, "_internal_database", None)
+    if internal is not None:
+        databases.append(internal)
+    return databases
+
+
+# Each callback is a plain generator function so it can be unit-tested
+# directly, without standing up an SDK provider and a metric reader.
+
+
+def observe_sql_thread_limit(options=None):
+    "Size of the shared read-query thread pool (the num_sql_threads setting)."
+    for ds in _live_instances():
+        if ds.executor is None:
+            # num_sql_threads=0 - queries run on the event loop, no pool.
+            continue
+        yield otel_metrics.Observation(ds.setting("num_sql_threads"), {})
+
+
+def observe_sql_thread_queue_depth(options=None):
+    """
+    Read queries waiting for a free thread in the shared pool.
+
+    This is the saturation signal: sustained above zero means requests are
+    queueing on num_sql_threads. `_work_queue` is a private attribute of
+    ThreadPoolExecutor, so its absence is tolerated rather than fatal - a
+    missing gauge is much better than a crashed collection cycle.
+    """
+    for ds in _live_instances():
+        if ds.executor is None:
+            continue
+        work_queue = getattr(ds.executor, "_work_queue", None)
+        if work_queue is None:
+            continue
+        yield otel_metrics.Observation(work_queue.qsize(), {})
+
+
+def observe_pending_queries(options=None):
+    """
+    Read queries submitted to the pool and not yet finished, per database.
+
+    Summed across databases and compared against the thread limit, this is the
+    utilisation half of the saturation picture. `len()` is deliberately taken
+    without `_pending_execute_futures_lock`: it is atomic, and taking a lock
+    held on the request path from the collection thread would let telemetry
+    add latency to queries.
+    """
+    for ds in _live_instances():
+        for db in _databases_of(ds):
+            yield otel_metrics.Observation(
+                len(db._pending_execute_futures), {DB_NAMESPACE: db.name}
+            )
+
+
+def observe_write_queue_depth(options=None):
+    """
+    Writes queued behind the single write thread, per database.
+
+    Every database serialises its writes through one thread, so this is
+    unbounded backpressure that no amount of num_sql_threads will relieve.
+    """
+    for ds in _live_instances():
+        for db in _databases_of(ds):
+            write_queue = db._write_queue
+            if write_queue is None:
+                # No write has ever been queued for this database.
+                continue
+            yield otel_metrics.Observation(
+                write_queue.qsize(), {DB_NAMESPACE: db.name}
+            )
+
+
+def observe_open_connections(options=None):
+    "Open SQLite file connections tracked for closing, per database."
+    for ds in _live_instances():
+        for db in _databases_of(ds):
+            yield otel_metrics.Observation(
+                len(db._all_file_connections), {DB_NAMESPACE: db.name}
+            )
+
+
+sql_thread_limit_gauge = meter.create_observable_gauge(
+    "datasette.sql.threads.limit",
+    callbacks=[observe_sql_thread_limit],
+    unit="{thread}",
+    description="Maximum concurrent read queries (the num_sql_threads setting)",
+)
+
+sql_thread_queue_depth_gauge = meter.create_observable_gauge(
+    "datasette.sql.threads.queue_depth",
+    callbacks=[observe_sql_thread_queue_depth],
+    unit="{query}",
+    description="Read queries waiting for a free thread in the shared SQL pool",
+)
+
+pending_queries_gauge = meter.create_observable_gauge(
+    "datasette.sql.queries.pending",
+    callbacks=[observe_pending_queries],
+    unit="{query}",
+    description="Read queries submitted to the pool and not yet complete",
+)
+
+write_queue_depth_gauge = meter.create_observable_gauge(
+    "datasette.write.queue_depth",
+    callbacks=[observe_write_queue_depth],
+    unit="{write}",
+    description="Writes queued behind a database's single write thread",
+)
+
+open_connections_gauge = meter.create_observable_gauge(
+    "datasette.connections.open",
+    callbacks=[observe_open_connections],
+    unit="{connection}",
+    description="Open SQLite file connections tracked for closing",
+)
diff --git a/demos/otel/README.md b/demos/otel/README.md
index 71b5d6e8..263011d7 100644
--- a/demos/otel/README.md
+++ b/demos/otel/README.md
@@ -1,16 +1,18 @@
 # OpenTelemetry demo
 
-Datasette core depends on `opentelemetry-api` only. It emits spans and nothing else — it never
-creates a `TracerProvider`, never configures an exporter, and never sets a sampler. With no SDK
-installed every span is a no-op and costs approximately nothing.
+Datasette core depends on `opentelemetry-api` only. It emits spans and metrics and nothing
+else — it never creates a `TracerProvider` or a `MeterProvider`, never configures an exporter,
+and never sets a sampler. With no SDK installed every span and every instrument is a no-op and
+costs approximately nothing.
 
-That means "turning tracing on" is entirely the job of whoever runs Datasette. This directory
-shows two ways to do it, **neither of which needs Docker**:
+That means "turning telemetry on" is entirely the job of whoever runs Datasette. This directory
+shows several ways to do it, **none of which needs Docker**:
 
 | | |
 |---|---|
 | `otlp_receiver.py` | A ~150 line pure-Python OTLP/HTTP receiver — a real protobuf export, summarized in your terminal |
 | `just jaeger` | The same export into Jaeger's own binary, for a real trace UI |
+| `metrics_demo.py` | Saturates the SQL thread pool and prints the gauges — the question spans cannot answer |
 
 Both listen for OTLP/HTTP on port 4318, so the Datasette side is identical — run one or the
 other, not both. The `Justfile` in this directory wraps every command below; bare `just` lists
@@ -97,6 +99,49 @@ request produces exactly two traces — the request trace (~67 spans, rooted at
 span, with every `db.query` nested inside it across thread boundaries) and the
 `datasette.startup` trace (~26 spans of catalog queries and connection warm-up).
 
+## 3. The question spans cannot answer
+
+```bash
+uv run python demos/otel/metrics_demo.py
+```
+
+A trace tells you a query took 170ms. It does not tell you that 130ms of that was spent waiting for
+one of only three threads — "how many threads are busy right now" is a level, not an event, so no
+span can carry it. That is what metrics are for.
+
+The script registers a SQL function that sleeps, fires 12 concurrent 40ms queries at a pool of 3
+threads, and samples the gauges while they are in flight. Real output:
+
+```
+num_sql_threads = 3, firing 12 concurrent 40ms queries
+
+wall clock                     : 170ms
+  if fully serialised          : 480ms
+  with 3 threads perfectly used : 160ms
+
+Peak values sampled while the queries were in flight:
+
+  datasette.sql.queries.pending  {db.namespace=demo} = 12
+  datasette.sql.threads.queue_depth = 9
+
+Final collection:
+
+  datasette.sql.threads.limit
+      3
+  db.client.operation.duration  {datasette.operation=read, db.namespace=demo, db.system=sqlite}
+      count=16 sum=1.2605s min=0.0001s max=0.1695s
+  datasette.write.queue_wait  {db.namespace=demo}
+      count=1 sum=0.0002s min=0.0002s max=0.0002s
+```
+
+`queue_depth = 9` is the whole point: 12 queries, 3 threads, 9 of them sitting in a queue. Sustained
+above zero in production means requests are backing up on `num_sql_threads`, and no amount of
+reading traces would have told you that.
+
+Note also `max=0.1695s` on the duration histogram against a query whose actual work is 40ms. The
+gap is queue time. The two numbers together — 170ms observed, 40ms of work — are what distinguishes
+"my queries are slow" from "my pool is too small".
+
 ## Notes on the environment variables
 
 - **`opentelemetry-instrument` is required.** Setting `OTEL_TRACES_EXPORTER` and running plain
diff --git a/demos/otel/metrics_demo.py b/demos/otel/metrics_demo.py
new file mode 100644
index 00000000..82e4d26b
--- /dev/null
+++ b/demos/otel/metrics_demo.py
@@ -0,0 +1,175 @@
+"""
+Saturate Datasette's SQL thread pool and print the metrics that show it.
+
+Run it with no arguments and no infrastructure:
+
+    uv run python demos/otel/metrics_demo.py
+
+It builds a small database in a temporary directory, registers a deliberately
+slow SQL function, installs an in-memory OpenTelemetry SDK metric reader, then
+fires more concurrent queries than there are threads in the pool while
+sampling the gauges in the background.
+
+The point is the question spans cannot answer. A trace tells you a query took
+170ms; it does not tell you that 130ms of that was spent waiting for one of
+only three threads, because "how many threads are busy right now" is a level
+rather than an event. That is what `datasette.sql.threads.queue_depth` reports.
+
+Datasette core never installs a `MeterProvider` - the block near the top of
+this file is doing the job that `opentelemetry-instrument` would normally do.
+Core only emits, and whoever runs Datasette decides where it goes.
+"""
+
+import asyncio
+import sqlite3
+import sys
+import tempfile
+import time
+from pathlib import Path
+
+try:
+    from opentelemetry import metrics as otel_metrics
+    from opentelemetry.sdk.metrics import MeterProvider
+    from opentelemetry.sdk.metrics.export import InMemoryMetricReader
+except ImportError:
+    sys.exit(
+        "This demo needs the OpenTelemetry SDK:\n"
+        "    pip install opentelemetry-sdk\n"
+        "(Datasette itself only depends on opentelemetry-api.)"
+    )
+
+reader = InMemoryMetricReader()
+otel_metrics.set_meter_provider(MeterProvider(metric_readers=[reader]))
+
+from datasette import hookimpl
+from datasette.app import Datasette
+from datasette.plugins import pm
+
+NUM_SQL_THREADS = 3
+CONCURRENT_QUERIES = 12
+QUERY_MS = 40
+
+
+class SlowQueryPlugin:
+    "Registers a SQL function that sleeps, so queries occupy a pool thread."
+
+    __name__ = "slow-query-plugin"
+
+    @hookimpl
+    def prepare_connection(self, conn):
+        conn.create_function("slow_ms", 1, lambda ms: time.sleep(ms / 1000) or ms)
+
+
+def build_database(path):
+    conn = sqlite3.connect(path)
+    conn.execute("create table t (id integer primary key)")
+    conn.executemany("insert into t (id) values (?)", [[i] for i in range(100)])
+    conn.commit()
+    conn.close()
+
+
+def collect():
+    "One collection cycle -> {(metric name, attributes tuple): data point}."
+    points = {}
+    data = reader.get_metrics_data()
+    if data is None:
+        return points
+    for resource_metrics in data.resource_metrics:
+        for scope_metrics in resource_metrics.scope_metrics:
+            for metric in scope_metrics.metrics:
+                for point in metric.data.data_points:
+                    key = (metric.name, tuple(sorted((point.attributes or {}).items())))
+                    points[key] = point
+    return points
+
+
+async def sample_during_load(stop, peaks):
+    "Poll the gauges while the load is running and keep the highest seen."
+    while not stop.is_set():
+        points = collect()
+        for name in (
+            "datasette.sql.threads.queue_depth",
+            "datasette.sql.queries.pending",
+        ):
+            for (metric_name, attributes), point in points.items():
+                if metric_name != name:
+                    continue
+                current = getattr(point, "value", 0)
+                key = (name, attributes)
+                if current > peaks.get(key, -1):
+                    peaks[key] = current
+        await asyncio.sleep(0.002)
+
+
+def format_attributes(attributes):
+    if not attributes:
+        return ""
+    return "  {" + ", ".join(f"{k}={v}" for k, v in attributes) + "}"
+
+
+async def main():
+    tmpdir = Path(tempfile.mkdtemp())
+    db_path = tmpdir / "demo.db"
+    build_database(db_path)
+
+    pm.register(SlowQueryPlugin(), name="slow-query-plugin")
+    ds = Datasette([str(db_path)], settings={"num_sql_threads": NUM_SQL_THREADS})
+    await ds.invoke_startup()
+    db = ds.get_database("demo")
+
+    # Warm up so connection setup and schema introspection do not land in the
+    # middle of the measurement.
+    await db.execute("select 1")
+
+    print(
+        f"num_sql_threads = {NUM_SQL_THREADS}, "
+        f"firing {CONCURRENT_QUERIES} concurrent {QUERY_MS}ms queries\n"
+    )
+
+    peaks = {}
+    stop = asyncio.Event()
+    sampler = asyncio.ensure_future(sample_during_load(stop, peaks))
+
+    started = time.perf_counter()
+    await asyncio.gather(
+        *[db.execute(f"select slow_ms({QUERY_MS})") for _ in range(CONCURRENT_QUERIES)]
+    )
+    elapsed_ms = (time.perf_counter() - started) * 1000
+    stop.set()
+    await sampler
+
+    # A write, so the write-queue metrics have something to report.
+    await db.execute_write("create table if not exists written (id integer)")
+
+    serial_ms = CONCURRENT_QUERIES * QUERY_MS
+    ideal_ms = serial_ms / NUM_SQL_THREADS
+    print(f"wall clock                     : {elapsed_ms:.0f}ms")
+    print(f"  if fully serialised          : {serial_ms}ms")
+    print(f"  with {NUM_SQL_THREADS} threads perfectly used : {ideal_ms:.0f}ms\n")
+
+    print("Peak values sampled while the queries were in flight:\n")
+    for (name, attributes), peak in sorted(peaks.items()):
+        print(f"  {name}{format_attributes(attributes)} = {peak}")
+
+    print("\nFinal collection:\n")
+    points = collect()
+    for (name, attributes), point in sorted(points.items()):
+        if hasattr(point, "value"):
+            rendered = str(point.value)
+        else:
+            rendered = (
+                f"count={point.count} sum={point.sum:.4f}s "
+                f"min={point.min:.4f}s max={point.max:.4f}s"
+            )
+        print(f"  {name}{format_attributes(attributes)}")
+        print(f"      {rendered}")
+
+    print(
+        "\nThe queue_depth peak is the number of queries that were sitting "
+        "\nwaiting for a thread. Raising num_sql_threads is what moves it."
+    )
+    ds.close()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
diff --git a/docs/changelog.rst b/docs/changelog.rst
index a3d39a63..a266dc0c 100644
--- a/docs/changelog.rst
+++ b/docs/changelog.rst
@@ -11,6 +11,7 @@ Unreleased
 
 - Datasette's database layer now emits `OpenTelemetry <https://opentelemetry.io/>`__ spans: one per query, covering the full round trip including time spent waiting for a SQL worker thread, plus separate child spans for the execution itself and for time spent in the write queue. Datasette core depends on ``opentelemetry-api`` only and never installs an SDK provider, an exporter or a sampler, so there is no effect and no measurable overhead unless tracing is switched on externally - normally with the standard ``opentelemetry-instrument`` agent. See :ref:`internals_telemetry`. (:issue:`1730`)
 - :ref:`db.execute(sql, ..., table=None) <database_execute>` has a new optional ``table=`` parameter, naming the table a query is about so it can be recorded on that query's OpenTelemetry span. It has no effect on query execution, and Datasette never derives it from the SQL. (:issue:`1730`)
+- Datasette core now also emits OpenTelemetry **metrics** covering SQL thread pool saturation, per-database write queue depth, open connections, query latency and time-limit interruptions. These answer operational questions that spans structurally cannot - "am I saturating my :ref:`setting_num_sql_threads` threads?" is a level, not an event - and they survive trace sampling. As with spans, core installs no ``MeterProvider``, so there is no cost unless metrics are collected externally. See :ref:`internals_telemetry`. (:issue:`1730`)
 - **Breaking change:** Datasette's hand-rolled tracer has been removed, now that OpenTelemetry covers the same ground. The ``?_trace=1`` query string parameter, the ``trace_debug`` setting and the ``datasette.tracer`` module are all gone. ``datasette.tracer.trace()`` and ``datasette.tracer.trace_child_tasks()`` were documented plugin APIs, so any plugin importing them will now raise ``ModuleNotFoundError`` and needs a new release. `datasette-pretty-traces <https://datasette.io/plugins/datasette-pretty-traces>`__ does not import that module, but it renders ``?_trace=1`` output, so it no longer has anything to display. (:issue:`1730`)
 
 .. _v1_0_a38:
diff --git a/tests/conftest.py b/tests/conftest.py
index d1d60ebf..5346721f 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -114,6 +114,106 @@ def otel_spans():
     yield _otel_span_exporter
 
 
+_otel_metric_reader = None
+
+
+@pytest.fixture(scope="session", autouse=True)
+def _otel_meter_provider():
+    """
+    Install a real OTel SDK MeterProvider + InMemoryMetricReader once per
+    process.
+
+    Unlike the tracer, ordering is not load-bearing here: `_ProxyMeter` and
+    the `_ProxyInstrument`s it hands out forward to a provider installed
+    *after* they were created, whereas `ProxyTracer` permanently caches the
+    first concrete tracer it resolves. This fixture is still session-scoped
+    and autouse for symmetry, and so that a single reader collects for the
+    whole run.
+
+    DELTA temporality is chosen for counters and histograms so that each
+    collection reports only what happened since the previous one. With the
+    SDK default of CUMULATIVE, every metrics test would see every query run
+    by every earlier test in the session.
+    """
+    global _otel_metric_reader
+    try:
+        from opentelemetry import metrics as otel_metrics
+        from opentelemetry.sdk.metrics import Counter, Histogram, MeterProvider
+        from opentelemetry.sdk.metrics.export import (
+            AggregationTemporality,
+            InMemoryMetricReader,
+        )
+    except ImportError:
+        return
+    reader = InMemoryMetricReader(
+        preferred_temporality={
+            Counter: AggregationTemporality.DELTA,
+            Histogram: AggregationTemporality.DELTA,
+        }
+    )
+    otel_metrics.set_meter_provider(MeterProvider(metric_readers=[reader]))
+    _otel_metric_reader = reader
+
+
+class MetricsCollector:
+    """
+    Thin reader over an `InMemoryMetricReader`.
+
+    `collect()` runs a collection cycle - which is what invokes the observable
+    gauge callbacks - and snapshots the result. Queries then run against that
+    snapshot rather than re-collecting, so a test that inspects several
+    metrics sees one consistent moment and does not drain delta state twice.
+    """
+
+    def __init__(self, reader):
+        self.reader = reader
+        self.snapshot = {}
+
+    def collect(self):
+        self.snapshot = {}
+        data = self.reader.get_metrics_data()
+        if data is None:
+            return self.snapshot
+        for resource_metrics in data.resource_metrics:
+            for scope_metrics in resource_metrics.scope_metrics:
+                for metric in scope_metrics.metrics:
+                    self.snapshot.setdefault(metric.name, []).extend(
+                        metric.data.data_points
+                    )
+        return self.snapshot
+
+    def points(self, name, attributes=None):
+        "Data points for `name` whose attributes are a superset of `attributes`."
+        found = []
+        for point in self.snapshot.get(name, []):
+            point_attributes = dict(point.attributes or {})
+            if all(point_attributes.get(k) == v for k, v in (attributes or {}).items()):
+                found.append(point)
+        return found
+
+    def point(self, name, attributes=None):
+        "The single matching data point, asserting there is exactly one."
+        found = self.points(name, attributes)
+        assert len(found) == 1, (
+            f"expected exactly one {name} point matching {attributes}, "
+            f"got {len(found)}: {found}"
+        )
+        return found[0]
+
+
+@pytest.fixture
+def otel_metrics():
+    """
+    Function-scoped metrics collector. Drains any delta state accumulated by
+    earlier tests before yielding, so counts start from zero.
+    """
+    pytest.importorskip("opentelemetry.sdk")
+    if _otel_metric_reader is None:
+        pytest.skip("OpenTelemetry SDK meter provider was not installed")
+    _otel_metric_reader.get_metrics_data()
+    yield MetricsCollector(_otel_metric_reader)
+
+
 @pytest.fixture
 def bare_ds():
     """
diff --git a/tests/test_telemetry_metrics.py b/tests/test_telemetry_metrics.py
new file mode 100644
index 00000000..6d1efd1f
--- /dev/null
+++ b/tests/test_telemetry_metrics.py
@@ -0,0 +1,323 @@
+"""
+Tests for the OpenTelemetry metrics Datasette core emits.
+
+Two layers are tested separately and deliberately:
+
+- The gauge callbacks are plain generator functions, so they are called
+  directly for exact-value assertions. Going through the SDK for those would
+  be unreliable: the pool gauges carry no attribute identifying which
+  Datasette produced them, and a pytest session has many live instances, so
+  the SDK's last-value aggregation would report whichever one happened to be
+  observed last.
+
+- The SDK pipeline (instrument -> reader -> data points) is tested through
+  the `otel_metrics` fixture, using metrics that carry `db.namespace` - a
+  uniquely named in-memory database is enough to isolate those from every
+  other instance alive in the session.
+"""
+
+import asyncio
+
+import pytest
+
+from datasette import telemetry
+from datasette.app import Datasette
+from datasette.database import Database
+from datasette.utils.sqlite import sqlite3
+
+pytestmark = pytest.mark.filterwarnings("ignore::ResourceWarning")
+
+
+def observations(callback, datasette=None):
+    """
+    Run a gauge callback, optionally keeping only observations produced by one
+    Datasette's databases. Returns a list of (attributes dict, value).
+    """
+    results = []
+    names = None
+    if datasette is not None:
+        names = {db.name for db in telemetry._databases_of(datasette)}
+    for observation in callback():
+        attributes = dict(observation.attributes or {})
+        namespace = attributes.get("db.namespace")
+        if names is not None and namespace is not None and namespace not in names:
+            continue
+        results.append((attributes, observation.value))
+    return results
+
+
+@pytest.fixture
+def metrics_ds():
+    "A Datasette with a distinctive thread count and a uniquely named database."
+    ds = Datasette(
+        memory=True,
+        settings={"num_sql_threads": 7},
+    )
+    ds.add_memory_database("metrics_test_db")
+    try:
+        yield ds
+    finally:
+        ds.close()
+
+
+@pytest.mark.asyncio
+async def test_sql_thread_limit_gauge_reports_num_sql_threads(metrics_ds):
+    values = [value for _, value in observations(telemetry.observe_sql_thread_limit)]
+    # Other instances are alive in this session, so assert membership rather
+    # than uniqueness - 7 is distinctive enough to only come from metrics_ds.
+    assert 7 in values
+
+
+@pytest.mark.asyncio
+async def test_no_thread_gauges_in_non_threaded_mode():
+    """
+    num_sql_threads=0 means there is no pool at all, so the pool gauges must
+    skip the instance rather than report a bogus limit of 0.
+
+    The pool gauges carry no attributes, so the assertion is that adding this
+    instance produces no *additional* observations.
+    """
+    before_limits = len(list(telemetry.observe_sql_thread_limit()))
+    before_depths = len(list(telemetry.observe_sql_thread_queue_depth()))
+    ds = Datasette(memory=True, settings={"num_sql_threads": 0})
+    try:
+        assert ds.executor is None
+        assert len(list(telemetry.observe_sql_thread_limit())) == before_limits
+        assert len(list(telemetry.observe_sql_thread_queue_depth())) == before_depths
+        # Per-database gauges are unaffected - they do not depend on the pool.
+        assert observations(telemetry.observe_pending_queries, ds)
+    finally:
+        ds.close()
+
+
+@pytest.mark.asyncio
+async def test_pending_queries_gauge_tracks_in_flight_queries(metrics_ds):
+    db = metrics_ds.get_database("metrics_test_db")
+    attributes = {"db.namespace": "metrics_test_db"}
+
+    def value():
+        points = [
+            v
+            for a, v in observations(telemetry.observe_pending_queries, metrics_ds)
+            if a == attributes
+        ]
+        assert len(points) == 1
+        return points[0]
+
+    assert value() == 0
+
+    # sqlite3.sleep is not a thing, so block the worker thread on an event we
+    # control from the event loop and sample the gauge while it is held.
+    release = asyncio.Event()
+    loop = asyncio.get_running_loop()
+    entered = asyncio.Event()
+
+    def blocking_fn(conn):
+        loop.call_soon_threadsafe(entered.set)
+        asyncio.run_coroutine_threadsafe(release.wait(), loop).result()
+        return "done"
+
+    task = asyncio.ensure_future(db.execute_fn(blocking_fn))
+    await entered.wait()
+    assert value() == 1, "a query occupying a pool thread must be counted as pending"
+    release.set()
+    assert await task == "done"
+    assert value() == 0, "the count must drop once the query completes"
+
+
+@pytest.mark.asyncio
+async def test_write_queue_depth_gauge(metrics_ds):
+    db = metrics_ds.get_database("metrics_test_db")
+    attributes = {"db.namespace": "metrics_test_db"}
+
+    def depths():
+        return [
+            v
+            for a, v in observations(telemetry.observe_write_queue_depth, metrics_ds)
+            if a == attributes
+        ]
+
+    # No write has ever been queued, so there is no queue and no observation -
+    # rather than a fabricated zero for a queue that does not exist.
+    assert depths() == []
+
+    await db.execute_write("create table t (id integer primary key)")
+    assert depths() == [0], "an idle write queue reports zero, not nothing"
+
+
+@pytest.mark.asyncio
+async def test_open_connections_gauge(metrics_ds, tmp_path):
+    path = str(tmp_path / "conns.db")
+    sqlite3.connect(path).execute("create table t (id integer primary key)")
+    db = metrics_ds.add_database(Database(metrics_ds, path=path), name="conns_db")
+    attributes = {"db.namespace": "conns_db"}
+
+    def open_connections():
+        points = [
+            v
+            for a, v in observations(telemetry.observe_open_connections, metrics_ds)
+            if a == attributes
+        ]
+        assert len(points) == 1
+        return points[0]
+
+    assert open_connections() == 0
+    await db.execute("select 1")
+    assert open_connections() >= 1, "executing a query opens a tracked file connection"
+
+
+@pytest.mark.asyncio
+async def test_operation_duration_histogram_read(otel_metrics):
+    ds = Datasette(memory=True)
+    ds.add_memory_database("duration_read_db")
+    try:
+        db = ds.get_database("duration_read_db")
+        await db.execute("select 1")
+        otel_metrics.collect()
+        point = otel_metrics.point(
+            "db.client.operation.duration",
+            {"db.namespace": "duration_read_db", "datasette.operation": "read"},
+        )
+        assert point.count == 1
+        assert point.sum > 0
+        assert dict(point.attributes)["db.system"] == "sqlite"
+        assert "error.type" not in dict(point.attributes)
+    finally:
+        ds.close()
+
+
+@pytest.mark.asyncio
+async def test_operation_duration_histogram_write(otel_metrics):
+    ds = Datasette(memory=True)
+    ds.add_memory_database("duration_write_db")
+    try:
+        db = ds.get_database("duration_write_db")
+        await db.execute_write("create table t (id integer primary key)")
+        otel_metrics.collect()
+        point = otel_metrics.point(
+            "db.client.operation.duration",
+            {"db.namespace": "duration_write_db", "datasette.operation": "write"},
+        )
+        assert point.count == 1
+        assert point.sum > 0
+    finally:
+        ds.close()
+
+
+@pytest.mark.asyncio
+async def test_operation_duration_records_error_type(otel_metrics):
+    "A failed query is still timed, and is separable from a successful one."
+    ds = Datasette(memory=True)
+    ds.add_memory_database("duration_error_db")
+    try:
+        db = ds.get_database("duration_error_db")
+        with pytest.raises(sqlite3.OperationalError):
+            await db.execute("select * from nope")
+        otel_metrics.collect()
+        point = otel_metrics.point(
+            "db.client.operation.duration",
+            {"db.namespace": "duration_error_db", "datasette.operation": "read"},
+        )
+        assert point.count == 1
+        assert dict(point.attributes)["error.type"] == "OperationalError"
+    finally:
+        ds.close()
+
+
+@pytest.mark.asyncio
+async def test_write_queue_wait_histogram(otel_metrics):
+    ds = Datasette(memory=True)
+    ds.add_memory_database("queue_wait_db")
+    try:
+        db = ds.get_database("queue_wait_db")
+        await db.execute_write("create table t (id integer primary key)")
+        await db.execute_write("insert into t (id) values (1)")
+        otel_metrics.collect()
+        point = otel_metrics.point(
+            "datasette.write.queue_wait", {"db.namespace": "queue_wait_db"}
+        )
+        assert point.count == 2, "one measurement per write dequeued"
+        assert point.sum >= 0
+    finally:
+        ds.close()
+
+
+@pytest.mark.asyncio
+async def test_interrupted_queries_counter(otel_metrics):
+    "The count of time-limit kills, which sampled traces cannot provide."
+    ds = Datasette(memory=True, settings={"sql_time_limit_ms": 1})
+    ds.add_memory_database("interrupted_db")
+    try:
+        db = ds.get_database("interrupted_db")
+        from datasette.database import QueryInterrupted
+
+        with pytest.raises(QueryInterrupted):
+            await db.execute("""
+                with recursive counter(x) as (
+                    select 0 union all select x + 1 from counter
+                )
+                select * from counter
+                """)
+        otel_metrics.collect()
+        point = otel_metrics.point(
+            "datasette.sql.queries.interrupted", {"db.namespace": "interrupted_db"}
+        )
+        assert point.value == 1
+    finally:
+        ds.close()
+
+
+@pytest.mark.asyncio
+async def test_metrics_are_reported_through_the_sdk_for_gauges(otel_metrics):
+    "End-to-end: a gauge callback reaches the reader as a data point."
+    ds = Datasette(memory=True)
+    ds.add_memory_database("gauge_pipeline_db")
+    try:
+        await ds.get_database("gauge_pipeline_db").execute("select 1")
+        otel_metrics.collect()
+        point = otel_metrics.point(
+            "datasette.sql.queries.pending", {"db.namespace": "gauge_pipeline_db"}
+        )
+        assert point.value == 0
+        assert otel_metrics.points("datasette.sql.threads.limit")
+    finally:
+        ds.close()
+
+
+def test_closed_datasette_stops_being_observed():
+    ds = Datasette(memory=True)
+    ds.add_memory_database("closed_db")
+    assert observations(telemetry.observe_pending_queries, ds)
+    ds.close()
+    names = [
+        attributes.get("db.namespace")
+        for attributes, _ in observations(telemetry.observe_pending_queries)
+    ]
+    assert "closed_db" not in names
+
+
+def test_registry_holds_instances_weakly():
+    """
+    Registering an instance must never be the thing that keeps it alive.
+
+    A stand-in object is used rather than a real Datasette because a Datasette
+    with a temp-disk internal database is pinned for the life of the process
+    by `Database.__init__`'s `atexit.register(self._cleanup_temp_file)`, which
+    holds the Database, which holds the Datasette. That is pre-existing and
+    unrelated to telemetry; what is tested here is that this registry adds no
+    reference of its own.
+    """
+    import gc
+    import weakref
+
+    class FakeDatasette:
+        pass
+
+    fake = FakeDatasette()
+    telemetry.register_datasette(fake)
+    assert fake in telemetry._live_instances()
+    ref = weakref.ref(fake)
+    del fake
+    gc.collect()
+    assert ref() is None
+    assert not any(isinstance(ds, FakeDatasette) for ds in telemetry._live_instances())