diff --git a/datasette/database.py b/datasette/database.py
index ec83ecad..96d42a21 100644
--- a/datasette/database.py
+++ b/datasette/database.py
@@ -404,22 +404,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()
@@ -449,9 +452,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()
@@ -741,7 +745,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()
diff --git a/datasette/telemetry_registry.py b/datasette/telemetry_registry.py
index 3c775a7b..e4082347 100644
--- a/datasette/telemetry_registry.py
+++ b/datasette/telemetry_registry.py
@@ -459,7 +459,9 @@ M_OPERATION_DURATION = MetricName(
     HISTOGRAM,
     "s",
     "Duration of a SQL operation. The standard OpenTelemetry semantic "
-    "convention metric, and the one that survives trace sampling.",
+    "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,
 )
diff --git a/docs/internals.rst b/docs/internals.rst
index 3b1ae285..b82ebfbc 100644
--- a/docs/internals.rst
+++ b/docs/internals.rst
@@ -2466,7 +2466,7 @@ This reference is generated from ``datasette/telemetry_registry.py``, like the s
 .. ]]]
 
 ``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.
+    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``.
 
diff --git a/tests/test_telemetry_metrics.py b/tests/test_telemetry_metrics.py
index f1ac4a56..5aade58b 100644
--- a/tests/test_telemetry_metrics.py
+++ b/tests/test_telemetry_metrics.py
@@ -460,3 +460,73 @@ def test_histograms_spread_values_across_buckets(
         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()