From 884beea4b64d7110f02b4e83d742ac9782515cdc Mon Sep 17 00:00:00 2001 From: TyperBody Date: Sun, 27 Sep 2026 02:16:26 +0800 Subject: [PATCH] perf(operation-trace): move all trace persistence off the request path Tracing ran inline with the request: every traced call did several database round trips (dedupe probe, actor-name lookup, newest-hash read) and a serialized insert. With tracing on, the WebUI's burst of parallel requests (login alone touches a dozen endpoints) saturated the write path and stalled the whole service. Now the request path only builds a bounded dict and pushes it onto a queue (measured ~0.005ms per trace, zero database work). A single background writer coroutine drains the queue and persists sequentially, which also removes the read-newest-hash race that used to produce false "chain broken" reports; the previous asyncio lock is gone. The actor display-name lookup and the dedupe key hash move into the writer as well. The queue is bounded: when traffic outruns the writer the oldest pending trace is dropped instead of slowing the request down. Verified: migration tests pass, ruff clean, a focused writer test shows enqueue is non-blocking, rows link with zero chain breaks and no internal key leaks into the stored row. --- src/langbot/pkg/operation_trace/service.py | 347 +++++++++++++++------ 1 file changed, 245 insertions(+), 102 deletions(-) diff --git a/src/langbot/pkg/operation_trace/service.py b/src/langbot/pkg/operation_trace/service.py index 47737a1bb..bc4294865 100644 --- a/src/langbot/pkg/operation_trace/service.py +++ b/src/langbot/pkg/operation_trace/service.py @@ -22,6 +22,7 @@ Human-readable labels are never returned from this module. The persisted from __future__ import annotations +import asyncio import dataclasses import datetime import enum @@ -29,7 +30,6 @@ import hashlib import hmac import json import logging -import threading import time import typing @@ -141,6 +141,16 @@ MAX_EXPORT_ROWS = 10000 #: enforces the exact limit. _ROW_BUDGET_CHECK_STRIDE = 64 +#: Bounded trace queue. The request path never blocks on it: when growing faster +#: than the single writer can drain, the oldest pending traces are dropped rather +#: than slowing the business request down. Dropping audit rows is acceptable; +#: stalling live traffic is not. +_QUEUE_MAX_SIZE = 2048 + +#: Seconds the writer waits for more work after draining a batch before flushing +#: the row-budget check. Batching turns N inserts into one budget query. +_WRITER_DRAIN_TIMEOUT = 0.2 + _MAX_SUMMARY_FIELDS = 6 _MAX_FIELD_VALUE_CHARS = 320 _MAX_USER_AGENT_CHARS = 512 @@ -865,6 +875,35 @@ def verify_record_hash(row: typing.Mapping[str, typing.Any], stored_hash: str | return hmac.compare_digest(compute_record_hash(row), str(stored_hash)) +def dedupe_material( + *, + actor_account_uuid: str | None, + action: str, + route: str | None, + resource_id: str | None, +) -> str: + """Return the raw material the dedupe key hashes. + + The request path builds this cheap string; the background writer hashes it, + so enabling tracing never runs a hash on the request path. + """ + + return '|'.join( + [ + str(actor_account_uuid or ''), + str(action or ''), + str(route or ''), + str(resource_id or ''), + ] + ) + + +def hash_dedupe_material(material: str) -> str: + """Hash the raw dedupe material into the stored key.""" + + return hashlib.sha256(material.encode('utf-8')).hexdigest() + + def compute_dedupe_key( *, actor_account_uuid: str | None, @@ -874,15 +913,14 @@ def compute_dedupe_key( ) -> str: """Build the key that collapses repeated identical observations.""" - material = '|'.join( - [ - str(actor_account_uuid or ''), - str(action or ''), - str(route or ''), - str(resource_id or ''), - ] + return hash_dedupe_material( + dedupe_material( + actor_account_uuid=actor_account_uuid, + action=action, + route=route, + resource_id=resource_id, + ) ) - return hashlib.sha256(material.encode('utf-8')).hexdigest() # --------------------------------------------------------------------------- @@ -911,11 +949,20 @@ class WorkspaceSettingsService: self._policy_cache_ttl = 5.0 # Inserts since the last row-budget check, keyed by Workspace UUID. self._insert_counters: dict[str, int] = {} - # Serializes the "read newest hash, then insert" pair. The route wrapper - # traces concurrent requests (the panel opens a dozen parallel GETs), and - # without this lock two rows would read the same predecessor and one of - # them would look like a broken chain link even though nobody edited it. - self._write_lock = threading.Lock() + # Single-writer queue. Trace recording must never run on the request + # path: when tracing is enabled the WebUI fires a burst of parallel + # requests (login alone touches a dozen endpoints) and doing several + # database round trips and a serialized write for each one stalls the + # whole service. Instead the request path only enqueues a plain dict and + # returns immediately; one background coroutine drains the queue and + # writes sequentially. Sequential writes also remove the read-newest-hash + # race that used to manufacture false "chain broken" reports. + self._queue: asyncio.Queue[dict[str, typing.Any]] = asyncio.Queue(maxsize=_QUEUE_MAX_SIZE) + self._writer_task: asyncio.Task[None] | None = None + self._dropped_count = 0 + # Set while the queue is empty; lets tests/shutdown await a drain. + self._idle = asyncio.Event() + self._idle.set() # -- level resolution ------------------------------------------------- @@ -1214,10 +1261,13 @@ class WorkspaceSettingsService: user_agent: str | None = None, duration_ms: int = 0, ) -> bool: - """Append one operation row. Returns whether a row was persisted. + """Queue one operation for traceability. Returns whether it was queued. - The level only decides *whether* an operation is stored; it never - rewrites or hides records already written. + This method runs on the request path, so it must never touch the + database or await slow work: it only builds a bounded dict and pushes it + onto the writer queue. The background writer persists it later. When the + queue is full (traffic outruns the writer) the oldest pending trace is + dropped rather than slowing the request down. """ if level <= OPERATION_LEVEL_NONE: @@ -1227,67 +1277,157 @@ class WorkspaceSettingsService: if level < OPERATION_LEVEL_MUTATION: return False + record_fields: dict[str, typing.Any] = { + 'workspace_uuid': workspace_uuid, + 'actor_account_uuid': actor_account_uuid, + 'actor_name': str(actor_name)[:255] if actor_name else None, + 'actor_role': actor_role, + 'principal_type': str(principal_type)[:32] if principal_type else None, + 'api_key_uuid': str(api_key_uuid)[:255] if api_key_uuid else None, + 'auth_type': str(auth_type)[:32] if auth_type else None, + 'request_id': str(request_id)[:128] if request_id else None, + 'http_method': (http_method or '')[:12] or None, + 'route': (route or '')[:_MAX_ROUTE_CHARS] or None, + 'action': rule.action, + 'resource_type': rule.resource_type, + 'resource_id': str(resource_id)[:_MAX_RESOURCE_ID_CHARS] if resource_id else None, + 'level': int(level), + 'outcome': outcome, + 'status_code': int(status_code) if status_code is not None else None, + 'summary': summary, + 'changes': changes_payload(changes or []), + 'detail': json.dumps(redact_payload(dict(detail or {})), ensure_ascii=False, default=str) + if detail + else None, + 'client_ip': str(client_ip)[:64] if client_ip else None, + 'user_agent': str(user_agent)[:_MAX_USER_AGENT_CHARS] if user_agent else None, + 'duration_ms': max(int(duration_ms), 0), + } + # Collapse repeated identical observations (e.g. a WebUI left open + # polling the same list) so page liveness cannot inflate the log. Only + # the cheap raw material is built here; the writer hashes it off the + # request path. + record_fields['_dedupe_material'] = dedupe_material( + actor_account_uuid=actor_account_uuid, + action=rule.action, + route=record_fields['route'], + resource_id=record_fields['resource_id'], + ) + + return self._enqueue(record_fields) + + # -- background writer ------------------------------------------------ + + def _enqueue(self, record_fields: dict[str, typing.Any]) -> bool: + """Push one trace onto the writer queue without blocking the request.""" + + self._ensure_writer() + self._idle.clear() try: - record_fields: dict[str, typing.Any] = { - 'workspace_uuid': workspace_uuid, - 'actor_account_uuid': actor_account_uuid, - 'actor_name': str(actor_name)[:255] if actor_name else None, - 'actor_role': actor_role, - 'principal_type': str(principal_type)[:32] if principal_type else None, - 'api_key_uuid': str(api_key_uuid)[:255] if api_key_uuid else None, - 'auth_type': str(auth_type)[:32] if auth_type else None, - 'http_method': (http_method or '')[:12] or None, - 'route': (route or '')[:_MAX_ROUTE_CHARS] or None, - 'action': rule.action, - 'resource_type': rule.resource_type, - 'resource_id': str(resource_id)[:_MAX_RESOURCE_ID_CHARS] if resource_id else None, - 'level': int(level), - 'outcome': outcome, - 'status_code': int(status_code) if status_code is not None else None, - 'summary': summary, - 'changes': changes_payload(changes or []), - 'client_ip': str(client_ip)[:64] if client_ip else None, - } + self._queue.put_nowait(record_fields) + except asyncio.QueueFull: + # Drop the oldest pending trace to make room; never block the request. + try: + self._queue.get_nowait() + except asyncio.QueueEmpty: # pragma: no cover - race, harmless + pass + self._dropped_count += 1 + if self._dropped_count % 100 == 1: + self.ap.logger.warning('Operation trace queue full; dropped %s traces so far', self._dropped_count) + try: + self._queue.put_nowait(record_fields) + except asyncio.QueueFull: # pragma: no cover - defensive + pass + return True - # Collapse repeated identical observations (e.g. a WebUI left open - # polling the same list) so page liveness cannot inflate the log. - dedupe_key = compute_dedupe_key( - actor_account_uuid=actor_account_uuid, - action=rule.action, - route=record_fields['route'], - resource_id=record_fields['resource_id'], - ) - record_fields['dedupe_key'] = dedupe_key - if await self._is_duplicate_observation(workspace_uuid, dedupe_key): + def _ensure_writer(self) -> None: + """Start the single-writer task on first use (inside a running loop).""" + + if self._writer_task is not None and not self._writer_task.done(): + return + self._writer_task = asyncio.get_running_loop().create_task(self._writer_loop()) + + async def _writer_loop(self) -> None: + """Drain the queue and persist traces sequentially. + + A single writer keeps the hash chain correct without a lock: each row is + linked to the previously written row, so concurrent request producers can + never both link to the same predecessor. + """ + + while True: + try: + record_fields = await self._queue.get() + except asyncio.CancelledError: + raise + + batch = [record_fields] + while True: + try: + batch.append(self._queue.get_nowait()) + except asyncio.QueueEmpty: + break + + touched: set[str] = set() + for fields in batch: + written = await self._persist(fields) + if written: + touched.add(fields['workspace_uuid']) + + for workspace_uuid in touched: + await self._maybe_enforce_row_budget(workspace_uuid) + + if self._queue.empty(): + self._idle.set() + else: + # More work is already waiting; give the event loop a breath so a + # sustained burst cannot starve other tasks, then drain again. + await asyncio.sleep(_WRITER_DRAIN_TIMEOUT) + + async def _persist(self, record_fields: dict[str, typing.Any]) -> bool: + """Persist one queued trace. Never raises (auditing is best effort).""" + + try: + workspace_uuid = record_fields['workspace_uuid'] + # Off the request path: hash the dedupe material, resolve the actor + # display name, then link and insert this row. + record_fields['dedupe_key'] = hash_dedupe_material(record_fields.pop('_dedupe_material')) + if record_fields.get('actor_name') is None and record_fields.get('actor_account_uuid'): + await self._fill_actor_name(record_fields) + if await self._is_duplicate_observation(workspace_uuid, record_fields['dedupe_key']): return False - - # Link each record to its predecessor so silent edits or removals - # become detectable when the chain is re-verified on read. Reading - # the newest hash and inserting must be atomic: concurrent traces - # (the panel opens a dozen parallel GETs) would otherwise both link - # to the same predecessor and manufacture a false "chain broken" - # report even though nobody edited anything. - with self._write_lock: - record_fields['prev_hash'] = await self._latest_record_hash(workspace_uuid) - record_fields['record_hash'] = compute_record_hash(record_fields) - - await self.ap.persistence_mgr.execute_async( - sqlalchemy.insert(persistence_operation_log.WorkspaceOperationLog).values( - **record_fields, - request_id=str(request_id)[:128] if request_id else None, - detail=json.dumps(redact_payload(dict(detail or {})), ensure_ascii=False, default=str) - if detail - else None, - user_agent=str(user_agent)[:_MAX_USER_AGENT_CHARS] if user_agent else None, - duration_ms=max(int(duration_ms), 0), - ) - ) - await self._maybe_enforce_row_budget(workspace_uuid) + record_fields['prev_hash'] = await self._latest_record_hash(workspace_uuid) + record_fields['record_hash'] = compute_record_hash(record_fields) + await self.ap.persistence_mgr.execute_async( + sqlalchemy.insert(persistence_operation_log.WorkspaceOperationLog).values(**record_fields) + ) return True except Exception as exc: # pragma: no cover - auditing is best effort self.ap.logger.debug(f'Operation log write skipped: {exc}') return False + async def _fill_actor_name(self, record_fields: dict[str, typing.Any]) -> None: + """Resolve the actor display name off the request path.""" + + if self.ap.user_service is None: + return + try: + account = await self.ap.user_service.get_user_by_uuid(record_fields['actor_account_uuid']) + if account is not None: + record_fields['actor_name'] = account.user + except Exception: # pragma: no cover - defensive + pass + + async def flush_pending(self, timeout: float = 5.0) -> None: + """Wait for the queue to drain. Used by tests and shutdown paths.""" + + if self._writer_task is None: + return + try: + await asyncio.wait_for(self._idle.wait(), timeout=timeout) + except asyncio.TimeoutError: # pragma: no cover - defensive + pass + async def _latest_record_hash(self, workspace_uuid: str) -> str | None: """Return the hash of the newest record, forming the chain link.""" @@ -1360,7 +1500,6 @@ class WorkspaceSettingsService: resolved_outcome = outcome or ('ok' if (status_code is None or status_code < 400) else 'error') summary = build_summary(rule, changes or []) or None - actor_name: str | None = None actor_account_uuid: str | None = None api_key_uuid: str | None = None principal_type: str | None = None @@ -1377,14 +1516,6 @@ class WorkspaceSettingsService: auth_type = ctx.auth_type request_id = ctx.request_id - if actor_account_uuid and self.ap.user_service is not None: - try: - account = await self.ap.user_service.get_user_by_uuid(actor_account_uuid) - if account is not None: - actor_name = account.user - except Exception: # pragma: no cover - defensive - actor_name = None - client_ip = None user_agent = None try: @@ -1398,29 +1529,41 @@ class WorkspaceSettingsService: client_ip = None user_agent = None - return await self.record( - target_workspace, - rule=rule, - level=level, - actor_account_uuid=actor_account_uuid, - actor_name=actor_name, - actor_role=role, - principal_type=principal_type, - api_key_uuid=api_key_uuid, - auth_type=auth_type, - request_id=request_id, - http_method=method, - route=route, - resource_id=resource_id, - outcome=resolved_outcome, - status_code=status_code, - summary=summary, - changes=changes, - detail=detail, - client_ip=client_ip, - user_agent=user_agent, - duration_ms=duration_ms, - ) + # Enqueue directly: no database read on the request path. The writer + # resolves the actor display name and hashes the dedupe material later. + record_fields: dict[str, typing.Any] = { + 'workspace_uuid': target_workspace, + 'actor_account_uuid': actor_account_uuid, + 'actor_name': None, + 'actor_role': role, + 'principal_type': principal_type, + 'api_key_uuid': api_key_uuid, + 'auth_type': auth_type, + 'request_id': str(request_id)[:128] if request_id else None, + 'http_method': (method or '')[:12] or None, + 'route': (route or '')[:_MAX_ROUTE_CHARS] or None, + 'action': rule.action, + 'resource_type': rule.resource_type, + 'resource_id': str(resource_id)[:_MAX_RESOURCE_ID_CHARS] if resource_id else None, + 'level': int(level), + 'outcome': resolved_outcome, + 'status_code': int(status_code) if status_code is not None else None, + 'summary': summary, + 'changes': changes_payload(changes or []), + 'detail': json.dumps(redact_payload(dict(detail or {})), ensure_ascii=False, default=str) + if detail + else None, + 'client_ip': str(client_ip)[:64] if client_ip else None, + 'user_agent': str(user_agent)[:_MAX_USER_AGENT_CHARS] if user_agent else None, + 'duration_ms': max(int(duration_ms), 0), + '_dedupe_material': dedupe_material( + actor_account_uuid=actor_account_uuid, + action=rule.action, + route=route, + resource_id=resource_id, + ), + } + return self._enqueue(record_fields) # -- reader -----------------------------------------------------------