Coverage for src/asgi_profiler/detectors/blocking.py: 100%

1"""Synchronous database calls made on the event loop.""" 

2 

3from __future__ import annotations 

4 

5from collections.abc import Iterator 

6 

7from ..models import Problem, Profile 

8from .settings import Settings 

9from .utils import fingerprint, total_span_ms 

10 

11TYPE = "blocking_db_call" 

12 

13 

14def detect(profile: Profile, settings: Settings) -> Iterator[Problem]: 

15 """Statements that stopped the event loop while they ran. 

16 

17 The mistake this catches is quiet and extremely common: a synchronous 

18 `Session` used inside an `async def` endpoint. Nothing raises, the tests 

19 pass, and the endpoint looks fine in isolation -- because in isolation it 

20 *is* fine. The damage only appears under concurrency, where every other 

21 request in flight is frozen for the duration of every query, and it 

22 surfaces as latency on endpoints that have nothing to do with the 

23 offending one. That is close to impossible to attribute from the outside, 

24 which is why nothing else reports it: a production tracer sees the symptom 

25 on the wrong endpoint and the cause on none. 

26 

27 `Query.blocking` is decided at capture time, where the thread and the 

28 greenlet are still visible -- see `instrument._blocks_the_loop`. What is 

29 left here is arithmetic. 

30 

31 **One problem per request, not one per call site.** The first version of 

32 this split by the innermost frame, and a single endpoint that blocked for 

33 30 ms reported two findings of 1 ms and 0 ms -- both true, neither the 

34 thing the reader needed, which is how long their event loop was actually 

35 frozen. The fix is the same wherever in the request it happens (an async 

36 engine, `run_in_threadpool`, or a `def` endpoint), so splitting the 

37 finding only splits the number that makes the case for doing it. 

38 """ 

39 offenders = [i for i, query in enumerate(profile.queries) if query.blocking] 

40 if not offenders: 

41 return 

42 queries = [profile.queries[i] for i in offenders] 

43 span_ms = total_span_ms(queries) 

44 if span_ms < settings.blocking_ms: 

45 return 

46 

47 # Attributed by total time rather than by count: one 200 ms report query 

48 # is the thing to fix, not the forty 1 ms lookups around it. 

49 sites: dict[str, float] = {} 

50 for query in queries: 

51 frame = query.stack[-1] if query.stack else "" 

52 sites[frame] = sites.get(frame, 0.0) + query.duration_ms 

53 worst = max(sites, key=lambda frame: sites[frame]) 

54 statements = "statement" if len(offenders) == 1 else "statements" 

55 

56 yield Problem( 

57 type=TYPE, 

58 fingerprint=fingerprint(TYPE, profile.group, worst), 

59 # Counted, not timed. The duration is in the detail and in `time_ms`; 

60 # leading with it would print "Blocking database call: 0.4 ms" for a 

61 # sync session on a development database -- which reads as trivia, and 

62 # is the exact case worth catching, because the same code against a 

63 # real database is a hundred times that number. 

64 title=f"Blocking database call: {len(offenders)} {statements} on the event loop", 

65 detail=_detail(len(offenders), span_ms, profile.duration_ms, worst, len(sites)), 

66 evidence={ 

67 "sql": queries[0].sql, 

68 "count": len(offenders), 

69 "span_ms": round(span_ms, 3), 

70 "frame": worst, 

71 "call_sites": len(sites), 

72 }, 

73 offenders=offenders[:50], 

74 time_ms=span_ms, 

75 ) 

76 

77 

78def _detail(count: int, span_ms: float, request_ms: float, frame: str, sites: int) -> str: 

79 statements = "statement" if count == 1 else "statements" 

80 where = f", mostly from {frame}" if frame else "" 

81 elsewhere = f" across {sites} call sites" if sites > 1 else "" 

82 share = f" ({span_ms / request_ms:.0%} of the request)" if request_ms else "" 

83 return ( 

84 f"{count} {statements}{elsewhere} ran synchronously on the event loop " 

85 f"thread{where}, holding it for {span_ms:.1f} ms{share}. Every other request " 

86 f"in flight waited that long. Use an async engine with `AsyncSession`, wrap " 

87 f"the call in `starlette.concurrency.run_in_threadpool`, or make the endpoint " 

88 f"`def` rather than `async def` -- Starlette already runs those in a worker " 

89 f"thread." 

90 )