diff --git a/datasette/database.py b/datasette/database.py index 22984d6f..7114a691 100644 --- a/datasette/database.py +++ b/datasette/database.py @@ -394,6 +394,9 @@ class Database: # Context raises "RuntimeError: cannot enter context ... already # entered". This propagates the caller's otel context (e.g. the # enclosing db.query span) onto the worker thread. + # + # It also propagates every *other* ContextVar - see the note in + # execute_fn() for why that is safe. ctx = contextvars.copy_context() return await asyncio.get_running_loop().run_in_executor( self.ds.executor, ctx.run, _run @@ -695,6 +698,22 @@ class Database: # Context raises "RuntimeError: cannot enter context ... # already entered". This propagates the caller's otel context # (e.g. the enclosing db.query span) onto the worker thread. + # + # copy_context() is not selective: it also carries Datasette's own + # ContextVars - _skip_permission_checks and _permission_check_cache + # (datasette/permissions.py), _in_datasette_client (app.py) and, + # until the hand-rolled tracer goes, trace_task_id (tracer.py) - + # into worker threads, where they previously took their defaults. + # That is safe, for two reasons. Nothing reads them on a worker + # thread: the permission code that reads the first two is async and + # only ever runs on the event loop. And Context.run() restores the + # thread's previous context when the callable returns, so a value + # cannot outlive the submit that carried it and reach the next task + # on this shared pool - "skip permission checks" in particular can + # never bleed from one request into another's query. Where a value + # would be read - a plugin calling datasette.in_client() or trace() + # from inside an execute_fn callable - seeing the submitting + # request's value is the more accurate answer, not a leak. ctx = contextvars.copy_context() future = self.ds.executor.submit(ctx.run, in_thread) self._pending_execute_futures.add(future) @@ -722,7 +741,19 @@ class Database: self._check_not_closed() page_size = page_size or self.ds.page_size time_limit_ms = self.ds.sql_time_limit_ms - if custom_time_limit and custom_time_limit < time_limit_ms: + # A caller that hands in a budget shorter than the instance-wide + # sql_time_limit_ms is saying "this may not finish, and that is an + # answer I can use" - and every such caller in core does treat the + # timeout as normal: table_counts() stores None per table, facet + # suggestion moves on to the next column, autocomplete falls back to a + # prefix query. Those timeouts are therefore not span errors. Without + # this, the homepage alone emits one red span per table (it counts + # every table under a 10ms budget) on every single hit. + # + # A query that runs out the instance-wide limit is a different event - + # nobody asked for a short budget, so it stays an error. + timeout_expected = bool(custom_time_limit) and custom_time_limit < time_limit_ms + if timeout_expected: time_limit_ms = custom_time_limit def sql_operation_in_thread(conn): @@ -733,38 +764,53 @@ class Database: # databases) - so it parents correctly to the enclosing # db.query span despite running on a different thread. # - # Callers passing log_sql_errors=False are probing and treat a - # failure as an expected answer - see the matching handling on the - # db.query span in execute(). Without this, facet suggestion marks - # two spans per text column as failed on every table page. + # Exception handling is explicit rather than left to the context + # manager's flags, which apply to every exception type alike. This + # span needs to tell two apart: an expected timeout is never an + # error, while a genuine SQL failure is one unless the caller + # passed log_sql_errors=False, meaning it was probing and treats + # failure as an expected answer. Without the latter, facet + # suggestion marks two spans per text column as failed on every + # table page; without the former, so does every homepage hit. with tracer.start_as_current_span( DB_QUERY_EXECUTE, - record_exception=log_sql_errors, - set_status_on_exception=log_sql_errors, - ): - with sqlite_timelimit(conn, time_limit_ms): - try: - cursor = conn.cursor() - cursor.execute(sql, params if params is not None else {}) - max_returned_rows = self.ds.max_returned_rows - if max_returned_rows == page_size: - max_returned_rows += 1 - if max_returned_rows and truncate: - rows = cursor.fetchmany(max_returned_rows + 1) - truncated = len(rows) > max_returned_rows - rows = rows[:max_returned_rows] - else: - rows = cursor.fetchall() - truncated = False - except (sqlite3.OperationalError, sqlite3.DatabaseError) as e: - if e.args == ("interrupted",): - raise QueryInterrupted(e, sql, params) - if log_sql_errors: - sys.stderr.write( - f"ERROR: conn={conn}, sql = {sql!r}, params = {params}: {e}\n" - ) - sys.stderr.flush() - raise + record_exception=False, + set_status_on_exception=False, + ) as execute_span: + try: + with sqlite_timelimit(conn, time_limit_ms): + try: + cursor = conn.cursor() + cursor.execute(sql, params if params is not None else {}) + max_returned_rows = self.ds.max_returned_rows + if max_returned_rows == page_size: + max_returned_rows += 1 + if max_returned_rows and truncate: + rows = cursor.fetchmany(max_returned_rows + 1) + truncated = len(rows) > max_returned_rows + rows = rows[:max_returned_rows] + else: + rows = cursor.fetchall() + truncated = False + except (sqlite3.OperationalError, sqlite3.DatabaseError) as e: + if e.args == ("interrupted",): + raise QueryInterrupted(e, sql, params) + if log_sql_errors: + sys.stderr.write( + f"ERROR: conn={conn}, sql = {sql!r}, params = {params}: {e}\n" + ) + sys.stderr.flush() + raise + except QueryInterrupted as e: + if not timeout_expected: + execute_span.record_exception(e) + execute_span.set_status(Status(StatusCode.ERROR, str(e))) + raise + except Exception as e: + if log_sql_errors: + execute_span.record_exception(e) + execute_span.set_status(Status(StatusCode.ERROR, str(e))) + raise if truncate: return Results(rows, truncated, cursor.description) @@ -801,9 +847,13 @@ class Database: try: results = await self.execute_fn(sql_operation_in_thread) except QueryInterrupted as e: - span.set_status(Status(StatusCode.ERROR, str(e))) + # datasette.interrupted is set either way - it is the + # signal worth having. Only the ERROR status is + # conditional; see the timeout_expected comment above. span.set_attribute(INTERRUPTED, True) - span.record_exception(e) + if not timeout_expected: + span.set_status(Status(StatusCode.ERROR, str(e))) + span.record_exception(e) raise except Exception as e: # log_sql_errors=False means the caller is probing and diff --git a/datasette/telemetry_registry.py b/datasette/telemetry_registry.py index 40b32cc5..2d0692a8 100644 --- a/datasette/telemetry_registry.py +++ b/datasette/telemetry_registry.py @@ -137,7 +137,11 @@ TRUNCATED = Attribute( INTERRUPTED = Attribute( "datasette.interrupted", "True if the query was cancelled for exceeding the time limit. The span " - "status is also set to ``ERROR``.", + "status is also set to ``ERROR``, unless the caller asked for a budget " + "shorter than :ref:`setting_sql_time_limit_ms` - as table counts, facet " + "suggestion and autocomplete all do - in which case running out of time " + "is an expected answer rather than a failure and the status is left " + "unset.", optional=True, ) SQL_ERROR_SUPPRESSED = Attribute( diff --git a/docs/internals.rst b/docs/internals.rst index b2215d9d..e7028637 100644 --- a/docs/internals.rst +++ b/docs/internals.rst @@ -2355,7 +2355,7 @@ Spans are ``SpanKind.INTERNAL`` unless a kind is listed below. Only ``db.query`` - ``datasette.time_limit_ms`` *(optional)* - The :ref:`setting_sql_time_limit_ms` value this query ran under. Set on reads, which are the queries that time limit applies to. - ``datasette.rows_returned`` *(optional)* - Number of rows a read returned. Set on the read path only, and only when the read succeeded. - ``datasette.truncated`` *(optional)* - True if the result was cut short by :ref:`setting_max_returned_rows`. - - ``datasette.interrupted`` *(optional)* - True if the query was cancelled for exceeding the time limit. The span status is also set to ``ERROR``. + - ``datasette.interrupted`` *(optional)* - True if the query was cancelled for exceeding the time limit. The span status is also set to ``ERROR``, unless the caller asked for a budget shorter than :ref:`setting_sql_time_limit_ms` - as table counts, facet suggestion and autocomplete all do - in which case running out of time is an expected answer rather than a failure and the status is left unset. - ``datasette.sql_error_suppressed`` *(optional)* - True when the query failed but the caller passed ``log_sql_errors=False``, meaning it was probing and treats failure as an expected answer. Facet suggestion does this against every column. - ``datasette.executescript`` *(optional)* - True for ``execute_write_script()``, which runs multiple statements. - ``datasette.executemany`` *(optional)* - True for ``execute_write_many()``, which runs one statement against many parameter sets. diff --git a/tests/test_telemetry.py b/tests/test_telemetry.py index 0a6d0192..0022883a 100644 --- a/tests/test_telemetry.py +++ b/tests/test_telemetry.py @@ -12,7 +12,7 @@ from opentelemetry import trace as otel_trace from opentelemetry.trace import SpanKind, StatusCode from datasette.app import Datasette -from datasette.database import Database +from datasette.database import Database, QueryInterrupted from datasette.telemetry import ( MAX_SQL_LENGTH, SCHEMA_URL, @@ -26,6 +26,15 @@ SECRET_PARAM_VALUE = "SUPER_SECRET_PARAM_VALUE_XYZ_123" INVALID_SQL = "select this_is_not_valid_sql from nowhere" +# Bounded so a broken time limit fails the test instead of hanging it, but far +# too long to finish inside any of the millisecond budgets used below. +SLOW_SQL = """ +with recursive counter(x) as ( + select 1 union all select x + 1 from counter where x < 50000000 +) +select max(x) from counter +""" + def _db_query_spans(otel_spans): return [span for span in otel_spans.get_finished_spans() if span.name == "db.query"] @@ -212,14 +221,22 @@ async def test_no_span_attribute_ever_contains_a_parameter_value(ds_client, otel @pytest.mark.asyncio -async def test_query_interrupted_sets_error_status(ds_client, otel_spans): - response = await ds_client.get( - "/fixtures/-/query.json", - params={"sql": "select sleep(0.05)", "_timelimit": 5}, - ) - assert response.status_code == 400 +async def test_query_interrupted_sets_error_status(otel_spans): + """ + A query that runs out the instance-wide sql_time_limit_ms is an error. - spans = _db_query_spans(otel_spans) + This used to force the timeout with `?_timelimit=5`, but a caller-supplied + budget shorter than the instance limit is now the signal that the timeout + was expected - see test_expected_timeout_is_not_a_span_error - so the + timeout has to come from the setting for this to still test what it was + written to test. + """ + ds = Datasette(memory=True, settings={"sql_time_limit_ms": 20}) + db = ds.add_memory_database("t09_instance_limit_timeout") + with pytest.raises(QueryInterrupted): + await db.execute(SLOW_SQL) + + spans = _spans_for_namespace(otel_spans, "t09_instance_limit_timeout") assert spans span = spans[-1] assert span.status.status_code == StatusCode.ERROR @@ -228,6 +245,93 @@ async def test_query_interrupted_sets_error_status(ds_client, otel_spans): assert all(event.name == "exception" for event in span.events) +async def _expected_timeout_count_span(otel_spans, database_name): + """ + Drive the real table_counts() path into a timeout; return its db.query span. + + table_counts() is where the headline instance of this lives: the homepage + counts every table under a 10ms budget and stores None for any table that + does not finish in time. Before this was fixed, a two-table database + produced four ERROR spans - two db.query and two db.query.execute - on + every single homepage hit. + """ + db = Datasette(memory=True).add_memory_database(database_name) + await db.execute_write("create table big (id integer primary key, t text)") + await db.execute_write_many( + "insert into big (t) values (?)", [["x" * 50] for _ in range(11000)] + ) + # count_limit caps the scan at 10001 rows, and below 20ms sqlite_timelimit() + # runs its progress handler on every VM instruction, so 1ms is not a close + # call - a scan of that size takes single-digit milliseconds at best. + counts = await db.table_counts(1) + assert counts == { + "big": None + }, "the count did not actually time out, so the rest of this test is vacuous" + + spans = [ + span + for span in _spans_for_namespace(otel_spans, database_name) + if "count(*)" in span.attributes["db.query.text"] + ] + assert len(spans) == 1 + return spans[0] + + +@pytest.mark.asyncio +async def test_expected_timeout_is_not_a_span_error(otel_spans): + span = await _expected_timeout_count_span(otel_spans, "t09_expected_timeout") + # The useful signal survives; only the red status goes away. + assert span.attributes["datasette.interrupted"] is True + assert span.status.status_code != StatusCode.ERROR + assert not [event for event in span.events if event.name == "exception"] + + +@pytest.mark.asyncio +async def test_expected_timeout_does_not_error_the_inner_execute_span(otel_spans): + """ + The same fix has to reach db.query.execute, which sets its own status. + + Half of the original bug lived here: the inner span passed + set_status_on_exception=log_sql_errors, and table_counts() leaves + log_sql_errors at its True default, so it went ERROR too. + """ + span = await _expected_timeout_count_span(otel_spans, "t09_expected_timeout_inner") + children = _children_named(otel_spans, "db.query.execute", span.context) + assert len(children) == 1 + child = children[0] + assert child.status.status_code != StatusCode.ERROR + assert not [event for event in child.events if event.name == "exception"] + + +@pytest.mark.asyncio +async def test_unexpected_timeout_is_still_a_span_error(otel_spans): + """ + A custom_time_limit *above* sql_time_limit_ms is not a short budget. + + This is the half of the rule that stops the fix collapsing into "never + report timeouts": the caller asked for 5 seconds, the instance overruled it + at 20ms, and nobody expected that. + """ + ds = Datasette(memory=True, settings={"sql_time_limit_ms": 20}) + db = ds.add_memory_database("t09_custom_limit_ignored") + with pytest.raises(QueryInterrupted): + await db.execute(SLOW_SQL, custom_time_limit=5000) + + spans = _spans_for_namespace(otel_spans, "t09_custom_limit_ignored") + assert spans + span = spans[-1] + # Proves the caller's larger budget really was discarded - otherwise this + # would be asserting on a query that ran under a 5s limit. + assert span.attributes["datasette.time_limit_ms"] == 20 + assert span.attributes["datasette.interrupted"] is True + assert span.status.status_code == StatusCode.ERROR + assert any(event.name == "exception" for event in span.events) + + children = _children_named(otel_spans, "db.query.execute", span.context) + assert len(children) == 1 + assert children[0].status.status_code == StatusCode.ERROR + + @pytest.mark.asyncio async def test_unsuppressed_sql_error_is_a_span_error(ds_client, otel_spans): db = ds_client.ds.get_database("fixtures")