diff --git a/datasette/app.py b/datasette/app.py
index 31633837..ea632701 100644
--- a/datasette/app.py
+++ b/datasette/app.py
@@ -53,8 +53,10 @@ from .telemetry import (
     TelemetryMiddleware,
     _in_datasette_client,
     clamp_http_method,
+    register_datasette,
     request_span,
     tracer,
+    unregister_datasette,
 )
 from .telemetry_registry import HTTP_ROUTE, STARTUP
 from .tokens import TokenInvalid
@@ -651,6 +653,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
@@ -990,6 +996,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 0ed7ef06..411ca570 100644
--- a/datasette/database.py
+++ b/datasette/database.py
@@ -17,7 +17,15 @@ 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 callback_name, sql_attribute, sql_operation_name, tracer
+from .telemetry import (
+    callback_name,
+    record_operation_duration,
+    record_query_interrupted,
+    record_write_queue_wait,
+    sql_attribute,
+    sql_operation_name,
+    tracer,
+)
 from .telemetry_registry import (
     CALLBACK,
     DB_NAMESPACE,
@@ -316,9 +324,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):
@@ -340,9 +349,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):
@@ -373,9 +383,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)
@@ -409,22 +420,25 @@ class Database:
             span.set_attribute(DB_SYSTEM, "sqlite")
             span.set_attribute(DB_NAMESPACE, self.name)
             span.set_attribute(CALLBACK, callback_name(fn))
-            if self.ds.executor is None:
-                # non-threaded mode
-                return _run()
-            if not write:
-                # Immutable database - no writes can ever occur, so there is
-                # no write queue to block; run against a fresh read-only
-                # connection. copy_context() carries the caller's otel context
-                # onto the worker thread - see the notes in _execute_fn() for
-                # why it must be a fresh copy per submit and why carrying
-                # every ContextVar is safe.
-                ctx = contextvars.copy_context()
-                return await asyncio.get_running_loop().run_in_executor(
-                    self.ds.executor, ctx.run, _run
-                )
-            # Threaded mode - send to write thread
-            return await self._send_to_write_thread(fn, isolated_connection=True)
+            # "write" when mutable because the call blocks the write queue;
+            # "read" when immutable, where it runs on the read pool.
+            with record_operation_duration(self.name, "write" if write else "read"):
+                if self.ds.executor is None:
+                    # non-threaded mode
+                    return _run()
+                if not write:
+                    # Immutable database - no writes can ever occur, so there
+                    # is no write queue to block; run against a fresh
+                    # read-only connection. copy_context() carries the
+                    # caller's otel context onto the worker thread - see the
+                    # notes in _execute_fn() for why it must be a fresh copy
+                    # per submit and why carrying every ContextVar is safe.
+                    ctx = contextvars.copy_context()
+                    return await asyncio.get_running_loop().run_in_executor(
+                        self.ds.executor, ctx.run, _run
+                    )
+                # Threaded mode - send to write thread
+                return await self._send_to_write_thread(fn, isolated_connection=True)
 
     async def analyze_sql(self, sql, params=None) -> SQLAnalysis:
         self._check_not_closed()
@@ -454,9 +468,10 @@ class Database:
             span.set_attribute(DB_SYSTEM, "sqlite")
             span.set_attribute(DB_NAMESPACE, self.name)
             span.set_attribute(CALLBACK, name)
-            return await self._execute_write_fn(
-                fn, block=block, transaction=transaction, request=request
-            )
+            with record_operation_duration(self.name, "write"):
+                return await self._execute_write_fn(
+                    fn, block=block, transaction=transaction, request=request
+                )
 
     async def _execute_write_fn(self, fn, block=True, transaction=True, request=None):
         self._check_not_closed()
@@ -665,11 +680,13 @@ 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.
@@ -751,7 +768,8 @@ class Database:
             # Default exception handling applies, unlike execute(): there is
             # no log_sql_errors=False probing caller and no expected-timeout
             # budget on this path, so a raised exception is an error.
-            return await self._execute_fn(fn_in_execute_span)
+            with record_operation_duration(self.name, "read"):
+                return await self._execute_fn(fn_in_execute_span)
 
     async def _execute_fn(self, fn):
         self._check_not_closed()
@@ -915,7 +933,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
@@ -924,6 +943,10 @@ class Database:
                     if not timeout_expected:
                         span.set_status(Status(StatusCode.ERROR, str(e)))
                         span.record_exception(e)
+                        # Expected timeouts (a caller that opted into a shorter
+                        # budget, like facet suggestion) are not counted - see
+                        # the M_QUERIES_INTERRUPTED registry entry for why.
+                        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 19462585..4159b2ee 100644
--- a/datasette/telemetry.py
+++ b/datasette/telemetry.py
@@ -2,31 +2,49 @@
 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
 `NonRecordingSpan`. That is not free - a table page emits ~100 spans -
 but end-to-end page benchmarks put the overhead below their own
 run-to-run variation. Installing an SDK provider is what costs
-something measurable.
+something measurable. Every metric instrument is likewise a no-op
+without a provider, and the observable-gauge callbacks are never
+invoked at all.
 """
 
 import contextvars
 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,
     INTERNAL_CLIENT,
+    M_CONNECTIONS_OPEN,
+    M_OPERATION_DURATION,
+    M_QUERIES_INTERRUPTED,
+    M_QUERIES_PENDING,
+    M_THREADS_LIMIT,
+    M_THREADS_QUEUE_DEPTH,
+    M_WRITE_QUEUE_DEPTH,
+    M_WRITE_QUEUE_WAIT,
+    OPERATION,
     SERVER_ADDRESS,
     URL_PATH,
     URL_SCHEME,
@@ -64,6 +82,7 @@ _in_datasette_client = contextvars.ContextVar("in_datasette_client", default=Fal
 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
 
@@ -90,11 +109,11 @@ def callback_name(fn) -> str:
 # fixed allowlist - deliberately not a parse.
 #
 # This runs against arbitrary user-supplied SQL (the `?sql=` query string,
-# canned queries, anything typed into the query editor), and the attribute is
-# a candidate dimension on a query-duration metric in a later phase. A metric
-# series is keyed by its attribute values, so echoing back an arbitrary first
-# token would let one visitor's typo mint a new, permanent series. The
-# allowlist bounds that at a fixed, small set regardless of what anyone sends.
+# canned queries, anything typed into the query editor), and the attribute
+# has to stay safe to use as a metric dimension. A metric series is keyed by
+# its attribute values, so echoing back an arbitrary first token would let
+# one visitor's typo mint a new, permanent series. The allowlist bounds that
+# at a fixed, small set regardless of what anyone sends.
 DB_OPERATION_ALLOWLIST = frozenset(
     {
         "SELECT",
@@ -359,3 +378,242 @@ class TelemetryMiddleware:
                     if status >= 500 and not escaped:
                         span.set_status(Status(StatusCode.ERROR))
                         span.set_attribute(ERROR_TYPE, str(status))
+
+
+# --- Metrics --------------------------------------------------------------
+#
+# Two shapes. Observable gauges - a callback the SDK invokes on its own
+# collection cycle, so a no-provider install never runs them - answer level
+# questions no span can, like "am I saturating my SQL threads right now".
+# Synchronous histograms/counters are recorded inline on the query path and
+# survive trace sampling: 1% of traces still means 100% of the latency
+# distribution. Why each metric exists is documented on its registry entry.
+#
+# 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,
+        OPERATION: operation,
+    }
+
+
+# Each instrument passes the SDK a short plain-text description; the registry
+# entry for the same metric carries a longer RST one for the generated docs
+# (it can use `:ref:` roles, which an exported description string cannot).
+
+sql_operation_duration = meter.create_histogram(
+    M_OPERATION_DURATION,
+    unit=M_OPERATION_DURATION.unit,
+    description="Duration of a SQL operation issued by Datasette",
+    explicit_bucket_boundaries_advisory=M_OPERATION_DURATION.buckets,
+)
+
+write_queue_wait = meter.create_histogram(
+    M_WRITE_QUEUE_WAIT,
+    unit=M_WRITE_QUEUE_WAIT.unit,
+    description=(
+        "Time a write spent queued behind the single write thread for its database"
+    ),
+    explicit_bucket_boundaries_advisory=M_WRITE_QUEUE_WAIT.buckets,
+)
+
+queries_interrupted = meter.create_counter(
+    M_QUERIES_INTERRUPTED,
+    unit=M_QUERIES_INTERRUPTED.unit,
+    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 connections tracked for closing, per database."
+    for ds in _live_instances():
+        for db in _databases_of(ds):
+            yield otel_metrics.Observation(
+                len(db._all_connections), {DB_NAMESPACE: db.name}
+            )
+
+
+sql_thread_limit_gauge = meter.create_observable_gauge(
+    M_THREADS_LIMIT,
+    callbacks=[observe_sql_thread_limit],
+    unit=M_THREADS_LIMIT.unit,
+    description="Maximum concurrent read queries (the num_sql_threads setting)",
+)
+
+sql_thread_queue_depth_gauge = meter.create_observable_gauge(
+    M_THREADS_QUEUE_DEPTH,
+    callbacks=[observe_sql_thread_queue_depth],
+    unit=M_THREADS_QUEUE_DEPTH.unit,
+    description="Read queries waiting for a free thread in the shared SQL pool",
+)
+
+pending_queries_gauge = meter.create_observable_gauge(
+    M_QUERIES_PENDING,
+    callbacks=[observe_pending_queries],
+    unit=M_QUERIES_PENDING.unit,
+    description="Read queries submitted to the pool and not yet complete",
+)
+
+write_queue_depth_gauge = meter.create_observable_gauge(
+    M_WRITE_QUEUE_DEPTH,
+    callbacks=[observe_write_queue_depth],
+    unit=M_WRITE_QUEUE_DEPTH.unit,
+    description="Writes queued behind a database's single write thread",
+)
+
+open_connections_gauge = meter.create_observable_gauge(
+    M_CONNECTIONS_OPEN,
+    callbacks=[observe_open_connections],
+    unit=M_CONNECTIONS_OPEN.unit,
+    description="Open SQLite connections tracked for closing",
+)
diff --git a/datasette/telemetry_registry.py b/datasette/telemetry_registry.py
index 44805892..d269ae93 100644
--- a/datasette/telemetry_registry.py
+++ b/datasette/telemetry_registry.py
@@ -79,6 +79,33 @@ class SpanName(str):
         return f"SpanName({str(self)!r})"
 
 
+class MetricName(str):
+    "A metric name, carrying its instrument kind, unit and attributes."
+
+    __slots__ = ("attributes", "buckets", "description", "kind", "unit")
+
+    def __new__(cls, name, kind, unit, description, attributes=(), buckets=None):
+        self = super().__new__(cls, name)
+        self.kind = kind
+        self.unit = unit
+        self.description = description
+        self.attributes = tuple(attributes)
+        # Explicit histogram bucket boundaries, for histograms only. Passed to
+        # create_histogram() as explicit_bucket_boundaries_advisory and
+        # published in the generated docs, since an operator writing a
+        # histogram_quantile() query needs to know them.
+        self.buckets = tuple(buckets) if buckets is not None else None
+        return self
+
+    def __repr__(self):
+        return f"MetricName({str(self)!r})"
+
+
+COUNTER = "Counter"
+HISTOGRAM = "Histogram"
+GAUGE = "Observable gauge"
+
+
 # --- Attributes -----------------------------------------------------------
 #
 # Shared attributes are defined once and referenced by every span that sets
@@ -157,6 +184,7 @@ ERROR_TYPE = Attribute(
 
 DB_SYSTEM = Attribute("db.system", "Always ``sqlite``.")
 DB_NAMESPACE = Attribute("db.namespace", "Name of the database being queried.")
+OPERATION = Attribute("datasette.operation", "``read`` or ``write``.")
 DB_QUERY_TEXT = Attribute(
     "db.query.text",
     "The SQL, truncated to 2048 characters. Never the parameter values. "
@@ -395,3 +423,115 @@ def attribute_allowed(span, emitted_key):
     if span is None:
         return False
     return emitted_key in span.attributes
+
+
+# --- Metrics --------------------------------------------------------------
+
+# Every duration histogram here is in seconds, and OpenTelemetry's default
+# bucket boundaries are tuned for milliseconds - their first non-zero boundary
+# is 5, so without explicit boundaries every SQLite query lands in the single
+# (0, 5] second bucket and every quantile query returns noise.
+#
+# These are the OpenTelemetry semantic conventions' recommended boundaries for
+# db.client.operation.duration, in seconds, plus 0.0001 and 0.0005 at the
+# bottom. The deviation is deliberate: those boundaries assume a network
+# database client, whereas SQLite is in-process and a large fraction of real
+# queries run in 30-80us, which would otherwise all pile into the first
+# bucket and be indistinguishable from each other.
+#
+# One shared list is used for every duration histogram rather than a tailored
+# list each, so that dashboards stay comparable and a queue wait can be read
+# against the query duration it delays. It already spans 100us to 10s, which
+# covers both a fast in-process read and a write queued behind contention.
+DURATION_BUCKETS = (0.0001, 0.0005, 0.001, 0.005, 0.01, 0.05, 0.1, 0.5, 1, 5, 10)
+
+M_OPERATION_DURATION = MetricName(
+    "db.client.operation.duration",
+    HISTOGRAM,
+    "s",
+    "Duration of a SQL operation. The standard OpenTelemetry semantic "
+    "convention metric, and the one that survives trace sampling. "
+    "Callback-style calls (``execute_fn()`` and friends) are counted "
+    "alongside the SQL-string methods.",
+    (DB_SYSTEM, DB_NAMESPACE, OPERATION, ERROR_TYPE),
+    buckets=DURATION_BUCKETS,
+)
+
+M_WRITE_QUEUE_WAIT = MetricName(
+    "datasette.write.queue_wait",
+    HISTOGRAM,
+    "s",
+    "Time each write waited in its database's write queue. The metric "
+    "counterpart of the ``db.write.queue_wait`` span.",
+    (DB_NAMESPACE,),
+    buckets=DURATION_BUCKETS,
+)
+
+M_QUERIES_INTERRUPTED = MetricName(
+    "datasette.sql.queries.interrupted",
+    COUNTER,
+    "{query}",
+    "Queries cancelled for exceeding :ref:`setting_sql_time_limit_ms`. Worth "
+    "alerting on: a rising rate means the limit is too tight or a table has "
+    "outgrown its queries. A caller that opted into a deliberately shorter "
+    "budget - facet suggestion, for example - is not counted, for the same "
+    "reason its timeout is not a span error.",
+    (DB_NAMESPACE,),
+)
+
+M_THREADS_LIMIT = MetricName(
+    "datasette.sql.threads.limit",
+    GAUGE,
+    "{thread}",
+    "Maximum concurrent read queries - the :ref:`setting_num_sql_threads` "
+    "value. Not reported when ``num_sql_threads`` is ``0``, since then queries "
+    "run on the event loop and there is no pool.",
+)
+
+M_THREADS_QUEUE_DEPTH = MetricName(
+    "datasette.sql.threads.queue_depth",
+    GAUGE,
+    "{query}",
+    "Read queries waiting for a free thread. **This is the saturation "
+    "signal** - sustained above zero means requests are queueing on "
+    "``num_sql_threads``.",
+)
+
+M_QUERIES_PENDING = MetricName(
+    "datasette.sql.queries.pending",
+    GAUGE,
+    "{query}",
+    "Read queries submitted to the pool and not yet complete. Summed across "
+    "databases and compared against the thread limit, this is pool "
+    "utilisation.",
+    (DB_NAMESPACE,),
+)
+
+M_WRITE_QUEUE_DEPTH = MetricName(
+    "datasette.write.queue_depth",
+    GAUGE,
+    "{write}",
+    "Writes queued behind a database's single write thread. Backpressure that "
+    "raising ``num_sql_threads`` cannot relieve. Not reported for a database "
+    "that has never been written to.",
+    (DB_NAMESPACE,),
+)
+
+M_CONNECTIONS_OPEN = MetricName(
+    "datasette.connections.open",
+    GAUGE,
+    "{connection}",
+    "Open SQLite connections currently tracked for closing.",
+    (DB_NAMESPACE,),
+)
+
+METRICS = (
+    M_OPERATION_DURATION,
+    M_WRITE_QUEUE_WAIT,
+    M_QUERIES_INTERRUPTED,
+    M_THREADS_LIMIT,
+    M_THREADS_QUEUE_DEPTH,
+    M_QUERIES_PENDING,
+    M_WRITE_QUEUE_DEPTH,
+    M_CONNECTIONS_OPEN,
+)
diff --git a/docs/changelog.rst b/docs/changelog.rst
index a7875c29..15e51025 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. Callback-style calls - :ref:`db.execute_fn() <database_execute_fn>`, :ref:`db.execute_write_fn() <database_execute_write_fn>` and ``db.execute_isolated_fn()``, the documented way for plugins to run arbitrary SQL - are covered too, carrying ``datasette.callback`` in place of the SQL text. 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`)
 - Every HTTP request now gets an OpenTelemetry ``SERVER`` span, named after the request method and matched route, carrying ``http.route``, the response status and W3C trace context extracted from inbound headers - so every database span has a request to belong to, and Datasette joins distributed traces started by a proxy or calling service. The query string is never recorded. See :ref:`internals_telemetry_requests`. (: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`)
 
 Nothing is removed by the OpenTelemetry work: the ``?_trace=1`` query string parameter, the ``trace_debug`` setting and the :ref:`internals_tracer` module all continue to work as before.
 
diff --git a/docs/internals.rst b/docs/internals.rst
index eee40425..17fd6b5b 100644
--- a/docs/internals.rst
+++ b/docs/internals.rst
@@ -2535,7 +2535,7 @@ A few things catch people out the first time:
 
 - **Always set** ``OTEL_SERVICE_NAME``. Without it the SDK's default resource reports a ``service.name`` of ``unknown_service``, and your traces will be filed under that instead of under a name you can search for.
 
-- **Setting** ``OTEL_METRICS_EXPORTER=none`` **and** ``OTEL_LOGS_EXPORTER=none`` is worth doing unless your backend accepts those signals too - ``opentelemetry-distro`` defaults every signal to OTLP, and a traces-only backend will reject the other two noisily. Datasette itself emits no metrics and no logs through OpenTelemetry.
+- **Setting** ``OTEL_LOGS_EXPORTER=none`` is worth doing unless your backend accepts logs too - ``opentelemetry-distro`` defaults every signal to OTLP, and a backend that does not take a signal will reject it noisily. Datasette emits no logs through OpenTelemetry; it does emit metrics (see :ref:`internals_telemetry_metrics`), so set ``OTEL_METRICS_EXPORTER=none`` only if your backend does not accept them.
 
 Span reference
 --------------
@@ -2621,6 +2621,115 @@ That is the route's compiled regular expression, not a prettified ``/{database}/
 
 .. [[[end]]]
 
+.. _internals_telemetry_metrics:
+
+Metric reference
+----------------
+
+Spans describe events; metrics describe levels and rates. "Am I saturating my :ref:`setting_num_sql_threads` threads right now?" cannot be answered by any span, because it is a level sampled at collection time - and it is usually the first thing worth knowing about a busy Datasette, since ``num_sql_threads`` defaults to ``3``. Metrics also survive trace sampling: an operator keeping 1% of traces still gets 100% of every histogram and counter below.
+
+As with spans, core emits these through the OpenTelemetry API only. Without a ``MeterProvider`` every instrument is a no-op, and the observable-gauge callbacks are never invoked at all, so an uninstrumented install pays nothing for them.
+
+Every duration histogram is in **seconds**, with explicit bucket boundaries chosen for an in-process database - OpenTelemetry's default boundaries are tuned for milliseconds and would file every SQLite query into a single bucket, making quantile queries meaningless. The boundaries are listed with each histogram because a ``histogram_quantile()`` query is only as good as the buckets underneath it.
+
+This reference is generated from ``datasette/telemetry_registry.py``, like the span reference above.
+
+.. [[[cog
+    from telemetry_doc import metrics
+    metrics(cog)
+.. ]]]
+
+``db.client.operation.duration``
+    Histogram, unit ``s``. Duration of a SQL operation. The standard OpenTelemetry semantic convention metric, and the one that survives trace sampling. Callback-style calls (``execute_fn()`` and friends) are counted alongside the SQL-string methods.
+
+    Bucket boundaries: ``0.0001``, ``0.0005``, ``0.001``, ``0.005``, ``0.01``, ``0.05``, ``0.1``, ``0.5``, ``1``, ``5``, ``10``.
+
+    Attributes:
+
+    - ``db.system`` - Always ``sqlite``.
+    - ``db.namespace`` - Name of the database being queried.
+    - ``datasette.operation`` - ``read`` or ``write``.
+    - ``error.type`` - Set when the request failed: the exception class name if one escaped the application, otherwise the status code as a string for a 5xx response. A 4xx does **not** set this and does not set an error status - per semantic conventions a client error is not a server span's failure.
+
+``datasette.write.queue_wait``
+    Histogram, unit ``s``. Time each write waited in its database's write queue. The metric counterpart of the ``db.write.queue_wait`` span.
+
+    Bucket boundaries: ``0.0001``, ``0.0005``, ``0.001``, ``0.005``, ``0.01``, ``0.05``, ``0.1``, ``0.5``, ``1``, ``5``, ``10``.
+
+    Attributes:
+
+    - ``db.namespace`` - Name of the database being queried.
+
+``datasette.sql.queries.interrupted``
+    Counter, unit ``{query}``. Queries cancelled for exceeding :ref:`setting_sql_time_limit_ms`. Worth alerting on: a rising rate means the limit is too tight or a table has outgrown its queries. A caller that opted into a deliberately shorter budget - facet suggestion, for example - is not counted, for the same reason its timeout is not a span error.
+
+    Attributes:
+
+    - ``db.namespace`` - Name of the database being queried.
+
+``datasette.sql.threads.limit``
+    Observable gauge, unit ``{thread}``. Maximum concurrent read queries - the :ref:`setting_num_sql_threads` value. Not reported when ``num_sql_threads`` is ``0``, since then queries run on the event loop and there is no pool.
+
+    No attributes.
+
+``datasette.sql.threads.queue_depth``
+    Observable gauge, unit ``{query}``. Read queries waiting for a free thread. **This is the saturation signal** - sustained above zero means requests are queueing on ``num_sql_threads``.
+
+    No attributes.
+
+``datasette.sql.queries.pending``
+    Observable gauge, unit ``{query}``. Read queries submitted to the pool and not yet complete. Summed across databases and compared against the thread limit, this is pool utilisation.
+
+    Attributes:
+
+    - ``db.namespace`` - Name of the database being queried.
+
+``datasette.write.queue_depth``
+    Observable gauge, unit ``{write}``. Writes queued behind a database's single write thread. Backpressure that raising ``num_sql_threads`` cannot relieve. Not reported for a database that has never been written to.
+
+    Attributes:
+
+    - ``db.namespace`` - Name of the database being queried.
+
+``datasette.connections.open``
+    Observable gauge, unit ``{connection}``. Open SQLite connections currently tracked for closing.
+
+    Attributes:
+
+    - ``db.namespace`` - Name of the database being queried.
+
+.. [[[end]]]
+
+Exemplars
+~~~~~~~~~
+
+An OpenTelemetry `exemplar <https://opentelemetry.io/docs/specs/otel/metrics/data-model/#exemplars>`__ attaches a trace ID and span ID to one sample backing a histogram measurement. Where a spike in ``db.client.operation.duration`` alone tells you "queries were slow sometime in this minute", the exemplar attached to one of the samples in that spike gives you the trace ID of an actual slow query to open.
+
+Datasette needs no configuration to produce these. Every histogram measurement on the query path is recorded while a span for that operation is active, and the OpenTelemetry SDK's default exemplar filter attaches the current trace ID and span ID to any measurement recorded inside a sampled span - this is SDK behaviour that Datasette's instrumentation does not need to opt into. Four queries of increasing cost, each inside its own span, produced one exemplar per query on ``db.client.operation.duration``:
+
+.. code-block:: text
+
+    db.client.operation.duration count=4
+      exemplars: 4
+        value=0.001564s trace_id=ddfaf45fd4e14913497d7efeac95f381 span_id=fd5792bdbb01e533
+        value=0.006320s trace_id=34aea775ade11a3c5f716695731000fe span_id=25ed9e29dd84dbee
+        value=0.045253s trace_id=a65cb58d1460a179f0d04046ff51ed0d span_id=7f34d6378c85d062
+        value=0.305240s trace_id=6089f4c515c221c0ca7bb53667b37ac8 span_id=0516f4a6641eaa0b
+
+Exemplars are kept per histogram bucket - the SDK's default reservoir for an explicit-bucket histogram holds one exemplar per bucket - so the bucket boundaries above decide how many distinct traces a metric can point at. The same four queries, run against an earlier set of bucket boundaries under which every one of them fell into a single ``(0, 5]`` second bucket, produced one exemplar instead of four:
+
+.. code-block:: text
+
+    db.client.operation.duration count=4
+      exemplars: 1
+        value=0.305349s trace_id=cd40f9af396ad1e1d71a8832d70ac84a span_id=8a8c288609731fea
+
+Correcting the bucket boundaries had a second effect beyond fixing the quantiles: it also multiplied the number of traces reachable from this metric, one to four for this workload.
+
+Which export path you use matters here. The OTLP exporter carries exemplars through unchanged. ``opentelemetry-exporter-prometheus`` (as of ``0.65b0``) does not: the string ``exemplar`` does not appear anywhere in its source, and an OpenMetrics scrape of the workload above through that exporter contained zero exemplar markers - even though ``prometheus-client`` (``0.26.0``), the library it depends on for OpenMetrics output, supports the syntax. If exemplars need to reach Prometheus, the path that works is an OTLP collector writing to Prometheus, not a plugin scraping through that exporter. On that path, the Prometheus server needs `--enable-feature=exemplar-storage <https://prometheus.io/docs/prometheus/latest/feature_flags/>`__, and the scrape itself must use the OpenMetrics exposition format - Prometheus's default text format has no syntax for exemplars at all. Grafana then needs the Prometheus data source's `exemplar configuration <https://grafana.com/docs/grafana/latest/datasources/prometheus/configure/>`__ (``exemplarTraceIdDestinations``) pointed at a tracing data source before it will draw an exemplar as a clickable point rather than an ordinary sample.
+
+An exemplar can only exist for a trace that was sampled. The SDK's default exemplar filter only records one when the measurement happens inside a sampled span, and produces no exemplar at all rather than a link to a trace that was never kept. Verified: with the tracer provider's sampler set to ``ALWAYS_OFF``, the same four-query workload produced ``exemplars: 0`` on every data point. At 1% head sampling, 99% of measurements contribute no exemplar - but every exemplar you do get is guaranteed to resolve to a trace that exists.
+
 .. _internals_telemetry_requests:
 
 Requests and inbound trace context
diff --git a/docs/telemetry_doc.py b/docs/telemetry_doc.py
index d0af5dbd..725cb968 100644
--- a/docs/telemetry_doc.py
+++ b/docs/telemetry_doc.py
@@ -34,3 +34,22 @@ def spans(cog):
         if span.kind != SpanKind.INTERNAL:
             cog.out(f"    Kind: ``{span.kind.name}``.\n\n")
         _attribute_lines(cog, span.attributes)
+
+
+def metrics(cog):
+    from datasette.telemetry_registry import METRICS
+
+    cog.out("\n")
+    for metric in METRICS:
+        cog.out(f"``{metric}``\n")
+        cog.out(f"    {metric.kind}, unit ``{metric.unit}``. {metric.description}\n\n")
+        if metric.buckets:
+            boundaries = ", ".join(f"``{boundary}``" for boundary in metric.buckets)
+            cog.out(f"    Bucket boundaries: {boundaries}.\n\n")
+        if metric.attributes:
+            cog.out("    Attributes:\n\n")
+            for attribute in metric.attributes:
+                cog.out(f"    - ``{attribute}`` - {attribute.description}\n")
+            cog.out("\n")
+        else:
+            cog.out("    No attributes.\n\n")
diff --git a/tests/conftest.py b/tests/conftest.py
index 9922bc85..e5292615 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -114,6 +114,104 @@ 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 - see the metrics
+    banner in `datasette/telemetry.py` for the `_ProxyMeter`-vs-`ProxyTracer`
+    difference. 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..5229c4c9
--- /dev/null
+++ b/tests/test_telemetry_metrics.py
@@ -0,0 +1,532 @@
+"""
+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 threading
+import weakref
+
+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 live-instance registry is
+    narrowed to just this instance for the assertion - counting global
+    observations instead would let an unrelated instance being garbage
+    collected mid-test shift the baseline.
+    """
+    ds = Datasette(memory=True, settings={"num_sql_threads": 0})
+    try:
+        assert ds.executor is None
+        original = telemetry._live_datasettes
+        telemetry._live_datasettes = weakref.WeakSet([ds])
+        try:
+            assert list(telemetry.observe_sql_thread_limit()) == []
+            assert list(telemetry.observe_sql_thread_queue_depth()) == []
+        finally:
+            telemetry._live_datasettes = original
+        # 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_thread_queue_depth_gauge_reports_saturation():
+    """
+    The headline alerting metric must actually read above zero when reads
+    queue behind num_sql_threads. This also pins the private
+    `ThreadPoolExecutor._work_queue` attribute the callback depends on: if a
+    stdlib rename ever removes it, this fails instead of the metric silently
+    vanishing (the callback tolerates its absence at collection time).
+    """
+    ds = Datasette(memory=True, settings={"num_sql_threads": 1})
+    db = ds.add_memory_database("metrics_saturation_db")
+    entered = threading.Event()
+    release = threading.Event()
+
+    def blocker(conn):
+        entered.set()
+        assert release.wait(timeout=10)
+        return 1
+
+    try:
+        first = asyncio.ensure_future(db.execute_fn(blocker))
+        # Wait until the blocker owns the pool's only thread.
+        await asyncio.get_running_loop().run_in_executor(None, entered.wait, 10)
+        second = asyncio.ensure_future(db.execute_fn(lambda conn: 2))
+        # The second submission lands in the executor's queue on the next
+        # event-loop turn; poll briefly rather than assume the timing.
+        depths = []
+        for _ in range(500):
+            depths = [
+                value
+                for _, value in observations(telemetry.observe_sql_thread_queue_depth)
+            ]
+            if any(value >= 1 for value in depths):
+                break
+            await asyncio.sleep(0.01)
+        assert any(value >= 1 for value in depths), depths
+        release.set()
+        assert await first == 1
+        assert await second == 2
+    finally:
+        release.set()
+        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 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_operation_duration_records_write_error_type(otel_metrics):
+    """
+    Same as the read-path error test, but the write wrappers time a different
+    code path - `execute_write_fn`, the write thread and its reply future -
+    so error propagation through them is pinned separately.
+    """
+    ds = Datasette(memory=True)
+    ds.add_memory_database("duration_write_error_db")
+    try:
+        db = ds.get_database("duration_write_error_db")
+        with pytest.raises(sqlite3.OperationalError):
+            await db.execute_write("insert into nope values (1)")
+        otel_metrics.collect()
+        point = otel_metrics.point(
+            "db.client.operation.duration",
+            {"db.namespace": "duration_write_error_db", "datasette.operation": "write"},
+        )
+        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())
+
+
+HISTOGRAM_PROBES = [
+    # (instrument attribute on telemetry, metric name, isolating attributes)
+    (
+        "sql_operation_duration",
+        "db.client.operation.duration",
+        {"db.namespace": "bucket_probe_operation"},
+    ),
+    (
+        "write_queue_wait",
+        "datasette.write.queue_wait",
+        {"db.namespace": "bucket_probe_queue_wait"},
+    ),
+]
+
+# One value inside each of six distinct registry buckets. Under OpenTelemetry's
+# default boundaries - [0, 5, 10, 25, ...], meant for milliseconds - the first
+# five of these all land in (0, 5] and only 7.0 lands elsewhere, so the
+# "occupies six buckets" assertion below fails if the advisory is ever dropped.
+SPREAD = [0.00005, 0.0003, 0.002, 0.03, 0.8, 7.0]
+
+
+@pytest.mark.parametrize(
+    "instrument_name,metric_name,attributes",
+    HISTOGRAM_PROBES,
+    ids=[metric for _, metric, _ in HISTOGRAM_PROBES],
+)
+def test_histograms_spread_values_across_buckets(
+    otel_metrics, instrument_name, metric_name, attributes
+):
+    """
+    The registry's boundaries reach the SDK, and a realistic spread of
+    seconds-scale durations occupies more than one bucket.
+
+    Recording onto the instrument directly rather than driving a workload is
+    deliberate: real durations here are all tens of microseconds and would
+    share a bucket no matter what the boundaries were, which is exactly the
+    situation this test exists to detect.
+
+    `explicit_bounds` is compared against the registry rather than against the
+    instrument's own configuration - the instrument is built *from* the
+    registry, so that comparison would be a value against itself. What is
+    checked here is that the advisory survived the trip through the SDK.
+    """
+    from datasette.telemetry_registry import METRICS
+
+    metric = next(m for m in METRICS if m == metric_name)
+    instrument = getattr(telemetry, instrument_name)
+    for value in SPREAD:
+        instrument.record(value, attributes)
+
+    otel_metrics.collect()
+    point = otel_metrics.point(metric_name, attributes)
+
+    assert (
+        tuple(point.explicit_bounds) == metric.buckets
+    ), "the registry's boundaries did not reach the SDK"
+    assert point.count == len(SPREAD)
+    occupied = [count for count in point.bucket_counts if count]
+    assert len(occupied) == len(SPREAD), (
+        f"expected each of {SPREAD} in its own bucket, got bucket counts "
+        f"{list(point.bucket_counts)} for bounds {list(point.explicit_bounds)}"
+    )
+
+
+@pytest.mark.asyncio
+async def test_operation_duration_histogram_records_execute_fn(otel_metrics):
+    "Callback-style reads land in the same histogram as SQL-string reads."
+    ds = Datasette(memory=True)
+    ds.add_memory_database("duration_fn_db")
+    try:
+        db = ds.get_database("duration_fn_db")
+
+        def read_one(conn):
+            return conn.execute("select 1").fetchone()[0]
+
+        assert await db.execute_fn(read_one) == 1
+        otel_metrics.collect()
+        point = otel_metrics.point(
+            "db.client.operation.duration",
+            {"db.namespace": "duration_fn_db", "datasette.operation": "read"},
+        )
+        assert point.count == 1
+        assert point.sum > 0
+    finally:
+        ds.close()
+
+
+@pytest.mark.asyncio
+async def test_operation_duration_histogram_records_execute_write_fn(otel_metrics):
+    "Callback-style writes - the JSON write API's whole diet - are counted too."
+    ds = Datasette(memory=True)
+    ds.add_memory_database("duration_write_fn_db")
+    try:
+        db = ds.get_database("duration_write_fn_db")
+
+        def create_table(conn):
+            conn.execute("create table t (id integer primary key)")
+
+        await db.execute_write_fn(create_table)
+        otel_metrics.collect()
+        point = otel_metrics.point(
+            "db.client.operation.duration",
+            {"db.namespace": "duration_write_fn_db", "datasette.operation": "write"},
+        )
+        assert point.count == 1
+        assert point.sum > 0
+    finally:
+        ds.close()
+
+
+@pytest.mark.asyncio
+async def test_operation_duration_records_callback_error_type(otel_metrics):
+    "A callback that raises is still timed, with error.type from the exception."
+    ds = Datasette(memory=True)
+    ds.add_memory_database("duration_fn_error_db")
+    try:
+        db = ds.get_database("duration_fn_error_db")
+
+        def boom(conn):
+            raise ValueError("callback failed")
+
+        with pytest.raises(ValueError):
+            await db.execute_fn(boom)
+        otel_metrics.collect()
+        point = otel_metrics.point(
+            "db.client.operation.duration",
+            {"db.namespace": "duration_fn_error_db", "datasette.operation": "read"},
+        )
+        assert point.count == 1
+        assert dict(point.attributes)["error.type"] == "ValueError"
+    finally:
+        ds.close()
diff --git a/tests/test_telemetry_registry.py b/tests/test_telemetry_registry.py
index 9abbfffc..762f7fc2 100644
--- a/tests/test_telemetry_registry.py
+++ b/tests/test_telemetry_registry.py
@@ -393,6 +393,29 @@ def test_registry_entries_are_usable_as_plain_strings():
     assert f"{reg.DB_QUERY}.execute" == "db.query.execute"
 
 
+def test_every_histogram_declares_bucket_boundaries():
+    """
+    Every histogram must carry explicit boundaries, and only histograms may.
+
+    OpenTelemetry's default boundaries start at 5 and are meant for
+    milliseconds, so a seconds-valued histogram that inherits them records
+    everything into one bucket. This is a registry self-consistency check, not
+    a check that the boundaries reached the SDK - for that see
+    `test_histograms_spread_values_across_buckets` in test_telemetry_metrics.py.
+    """
+    for metric in reg.METRICS:
+        if metric.kind == reg.HISTOGRAM:
+            assert metric.buckets, f"{metric} is a histogram with no boundaries"
+            assert list(metric.buckets) == sorted(
+                set(metric.buckets)
+            ), f"{metric} boundaries must be ascending and unique"
+            assert metric.buckets[0] > 0, f"{metric} has a non-positive boundary"
+        else:
+            assert (
+                metric.buckets is None
+            ), f"{metric} is a {metric.kind} and cannot have bucket boundaries"
+
+
 def test_dynamic_span_lookup():
     """
     `dynamic=True` matching, which is how the request span resolves.
@@ -420,3 +443,132 @@ def test_span_and_attribute_lookup():
     assert not reg.attribute_allowed(reg.DB_QUERY, "db.namespace.extra")
     assert not reg.attribute_allowed(reg.DB_QUERY, "datasette.isolated_connection")
     assert not reg.attribute_allowed(None, "db.namespace")
+
+
+# --- Metric conformance ----------------------------------------------------
+
+
+@pytest_asyncio.fixture
+async def emitted_metrics(otel_metrics):
+    """
+    Every (metric name, attribute key) pair produced by a broad workload,
+    plus the raw set of metric names - the metric-side counterpart of the
+    `emitted` span fixture above.
+
+    Metrics use DELTA temporality (see `_otel_meter_provider`), and the
+    function-scoped `otel_metrics` fixture drains any state left by an
+    earlier test before yielding, so this collection is not polluted by
+    other tests in the session - only by other *instances*, which is why the
+    checks below key everything off attribute names rather than values.
+    """
+    # The span workload already reaches every synchronous metric except the
+    # interrupted counter: reads and writes drive db.client.operation.duration
+    # and datasette.write.queue_wait, and both the suppressed-error probe and
+    # the custom_time_limit interrupt raise through record_operation_duration,
+    # setting error.type.
+    ds = await exercise()
+
+    # datasette.sql.queries.interrupted counts only queries that exceed the
+    # *configured* limit - a caller opting into a deliberately short budget
+    # via custom_time_limit (as exercise() does) is excluded by design. So a
+    # second instance whose configured limit is tiny provides the real thing.
+    slow_name = _unique("registry_metrics_slow")
+    slow = Datasette(memory=True, settings={"sql_time_limit_ms": 5})
+    slow.add_memory_database(slow_name)
+    await slow.invoke_startup()
+    slow_db = slow.get_database(slow_name)
+    with pytest.raises(QueryInterrupted):
+        await slow_db.execute(
+            "with recursive c(x) as (select 0 union all select x+1 from c) "
+            "select * from c"
+        )
+
+    # Collect while both instances are still registered, so the observable
+    # gauges - which observe live instances at collection time - report.
+    otel_metrics.collect()
+    snapshot = otel_metrics.snapshot
+    assert snapshot, "no metrics captured - the fixture is not exercising anything"
+    pairs = set()
+    for metric_name, points in snapshot.items():
+        for point in points:
+            for key in point.attributes or {}:
+                pairs.add((metric_name, key))
+    ds.close()
+    slow.close()
+    return {"names": set(snapshot), "pairs": pairs}
+
+
+@pytest.mark.asyncio
+async def test_every_registered_metric_is_emitted(emitted_metrics):
+    "The both-ways name check for metrics."
+    names = emitted_metrics["names"]
+    missing = sorted(str(m) for m in reg.METRICS if m not in names)
+    assert not missing, f"documented but never emitted: {missing}"
+
+    unregistered = sorted(
+        name for name in names if name not in {str(m) for m in reg.METRICS}
+    )
+    assert not unregistered, f"emitted but not registered: {unregistered}"
+
+
+@pytest.mark.asyncio
+async def test_every_emitted_metric_attribute_is_registered(emitted_metrics):
+    """
+    An attribute added to a metric without a registry entry would be missing
+    from the docs - the metric-side counterpart of
+    `test_every_emitted_attribute_is_registered`.
+    """
+    metric_for = {str(m): m for m in reg.METRICS}
+    unregistered = sorted(
+        f"{metric_name} -> {key}"
+        for metric_name, key in emitted_metrics["pairs"]
+        # A metric name with no registry entry at all is already reported by
+        # test_every_registered_metric_is_emitted; do not double-report it
+        # here, and do not crash attribute_allowed() on a None metric.
+        if metric_name in metric_for
+        and not reg.attribute_allowed(metric_for[metric_name], key)
+    )
+    assert (
+        not unregistered
+    ), "these metric attributes are emitted but not registered: " + "\n".join(
+        unregistered
+    )
+
+
+@pytest.mark.asyncio
+async def test_every_registered_metric_attribute_is_emitted(emitted_metrics):
+    """
+    The direction nothing else catches: the docs must not describe a metric
+    attribute that no longer exists.
+
+    Unlike the span-side attribute check, this does not skip `optional`
+    attributes. The only optional metric attribute is `error.type` on
+    `db.client.operation.duration`, and the workload reaches it from two
+    independent directions: the suppressed-error probe and the
+    custom_time_limit interrupt in `exercise()`, both of which raise through
+    `record_operation_duration`. So it is checked like any other attribute
+    rather than exempted; marking something optional here would opt it out of
+    verification entirely.
+
+    Gauges with no registered attributes (`datasette.sql.threads.limit` and
+    `.queue_depth`) fall out correctly with no special case: their
+    `metric.attributes` is empty, so the inner loop makes no assertion.
+    """
+    emitted_keys_by_metric = {}
+    for metric_name, key in emitted_metrics["pairs"]:
+        emitted_keys_by_metric.setdefault(metric_name, set()).add(key)
+
+    missing = []
+    for metric in reg.METRICS:
+        if str(metric) not in emitted_metrics["names"]:
+            # Not emitted at all - already reported by
+            # test_every_registered_metric_is_emitted; do not double-report.
+            continue
+        emitted_keys = emitted_keys_by_metric.get(str(metric), set())
+        for attribute in metric.attributes:
+            if attribute not in emitted_keys:
+                missing.append(f"{metric} -> {attribute}")
+    assert not missing, (
+        "these metric attributes are documented but never emitted by the "
+        "test workload: " + ", ".join(sorted(missing))
+    )