From ec94728385f0ac08bf78cd384638903fd16bd1a4 Mon Sep 17 00:00:00 2001 From: Dmitry Prudnikov Date: Tue, 1 Sep 2026 23:54:15 +0300 Subject: [PATCH 01/10] feat(client): report where a query was written The server's advisor groups what it measures by the shape of the query, so its report can name a statement but not the place that wrote it: one line of Cypher usually has a dozen call sites. Built with debug_source_tracking, the client now sends the file, line and function each query was written at, and the report names the line instead. app_name and app_version ride along, naming the service when several share a database. The keys are the contract the Rust driver already speaks, and this fills the one it has to leave empty: it reads the caller through #[track_caller], and a Location carries no function name, while a Python frame does. Off by default and free while off: the flag gates the frame read itself, and the metadata argument is not passed at all, so a query makes the call it made before this existed. The frame is read directly rather than through inspect.stack, which walks the whole stack and opens each frame's source file to quote lines around it: file I/O per frame to answer a question about one of them. The synchronous client reads the location on the way in rather than inside the coroutine, because it hands that coroutine to an event loop and the frame that called it has returned by the time it runs. A query handed to create_task is past saving that way, so it reports nothing: what is left on the stack is an event-loop frame, and the advisor would collect the queries of every unrelated task under that one line, which is worse than silence. --- README.md | 28 +++ coordinode/coordinode/_source.py | 140 +++++++++++++ coordinode/coordinode/client.py | 118 ++++++++++- tests/unit/test_source_tracking.py | 325 +++++++++++++++++++++++++++++ 4 files changed, 607 insertions(+), 4 deletions(-) create mode 100644 coordinode/coordinode/_source.py create mode 100644 tests/unit/test_source_tracking.py diff --git a/README.md b/README.md index b071ba8..fc3df95 100644 --- a/README.md +++ b/README.md @@ -140,6 +140,34 @@ Three constraints are worth knowing before holding a transaction open: calls on it say so, and `rollback()` raises instead of promising a discard. Verify the data rather than blindly retrying, which can duplicate the writes. +## Finding the Slow Query's Author + +The server's query advisor groups what it measures by the shape of the query, +which is why its report can name a statement but not the place that wrote it: +one line of Cypher usually has a dozen call sites. Turn on source tracking and +each query carries the file, line and function it was written at, so the report +names the line instead. + +```python +client = CoordinodeClient( + "localhost:7080", + debug_source_tracking=True, + app_name="feed-service", # optional, for when services share a database + app_version="2.1.0", +) +``` + +It is off by default and free while off: no frame is read and the request goes +out exactly as it would have. Turn it on where you are looking rather than +everywhere, because what it sends is the paths of your source files. + +One case reports nothing rather than guessing. A query handed to +`asyncio.create_task` runs after the frame that created it has returned, so its +call site is genuinely gone by then; the alternative would be reporting an +event-loop line, which the advisor would fill with the queries of every +unrelated task that took the same path. `await client.cypher(...)` and the +synchronous client both report normally. + ## LangChain — GraphRAG Pipeline ```python diff --git a/coordinode/coordinode/_source.py b/coordinode/coordinode/_source.py new file mode 100644 index 0000000..8a88acd --- /dev/null +++ b/coordinode/coordinode/_source.py @@ -0,0 +1,140 @@ +"""Call-site attribution for queries. + +When a client is built with ``debug_source_tracking=True``, every query it +sends carries the source location it was written at. The server's query +advisor groups statistics by query shape, and this is what lets it name the +line that wrote the slow one rather than only the query text, which is +usually identical across a dozen call sites. + +Off by default, and off means untouched: no frame is read, no metadata is +built, nothing is sent. The cost of the feature is paid only where somebody +asked for it. + +The metadata keys are the wire contract shared with the Rust driver: + +===================== ========================================== +``x-source-file`` path of the file the call was written in +``x-source-line`` line number, as a string +``x-source-function`` qualified name of the enclosing function +``x-source-app`` application name, when one was configured +``x-source-version`` application version, when one was configured +===================== ========================================== + +``x-source-function`` is the one key the Rust driver leaves empty: it reads +the caller through ``#[track_caller]``, and a ``Location`` there carries no +function name. A Python frame does, so this driver fills it in. +""" + +from __future__ import annotations + +import asyncio +import os +import sys +from types import FrameType +from typing import NamedTuple + +# Directories whose frames are never a call site. This package is the obvious +# one; asyncio is here because a query started with create_task runs its body +# after the frame that started it has gone, leaving an event-loop frame in its +# place. See is_call_site. +# The separator is part of each prefix so that a directory merely NAMED like +# one of these ("coordinode-extra" beside "coordinode") is not swallowed too. +_NOT_CALL_SITES = ( + os.path.dirname(os.path.abspath(__file__)) + os.sep, + os.path.dirname(os.path.abspath(asyncio.__file__)) + os.sep, +) + + +class SourceLocation(NamedTuple): + """Where a query was written: the file, line and enclosing function.""" + + file: str + line: int + function: str + + +def is_call_site(location: SourceLocation) -> bool: + """Whether *location* is somewhere a person could have written the query. + + A location inside this package or inside asyncio is not: it is what is + left on the stack when the frame that wrote the query is already gone, + which happens to a query handed to ``create_task`` and awaited later. + Reporting it would not merely be useless. The advisor groups its + statistics by call site, so one event-loop line would collect the queries + of every unrelated task that took that path and present them as one place + in the code, which is worse than the feature being quiet. + """ + # A frame's filename is normally already absolute, and making one absolute + # is not free: for a relative path it asks the OS for the working + # directory, which would be a syscall on every query. + path = location.file if os.path.isabs(location.file) else os.path.abspath(location.file) + return not path.startswith(_NOT_CALL_SITES) + + +def capture(levels_up: int) -> SourceLocation | None: + """Read the frame *levels_up* above this function's caller. + + Reading one frame directly is what keeps this cheap. The alternative, + ``inspect.stack()``, walks the whole stack and opens each frame's source + file to quote the lines around it: a file system round trip per frame, to + answer a question about one of them. + + ``None`` comes back when the frame cannot be had — an interpreter with no + Python-level frame support, or a stack shorter than the walk. Tracking + goes quiet rather than failing a query over a debugging aid. + """ + getframe = getattr(sys, "_getframe", None) + if getframe is None: + return None + try: + # +1 for this frame, which the caller counts from rather than into. + frame: FrameType = getframe(levels_up + 1) + except ValueError: + # Asked for more stack than exists. + return None + code = frame.f_code + return SourceLocation( + file=code.co_filename, + line=frame.f_lineno, + # Qualified, so a method reads as "Class.method" rather than a bare + # name that says nothing about which class it belongs to. + function=code.co_qualname, + ) + + +def identity(app_name: str, app_version: str) -> tuple[tuple[str, str], ...]: + """The metadata naming the application, built once per client. + + It cannot change after the client is constructed, so building it per query + would be the same work repeated for every statement the client ever sends. + + Either part may be left out, and an empty one gets no header: the server + reads a missing key and an empty value the same way, so the header would + carry nothing. + """ + pairs = [] + if app_name: + pairs.append(("x-source-app", app_name)) + if app_version: + pairs.append(("x-source-version", app_version)) + return tuple(pairs) + + +def to_metadata( + location: SourceLocation | None, + app_identity: tuple[tuple[str, str], ...], +) -> tuple[tuple[str, str], ...]: + """Build the gRPC metadata for *location*, empty when there is none. + + The application identity rides along with the location rather than on its + own: the server reads them as one source context and discards the whole + context when the file is missing, so sending the identity alone would put + it on the wire for nothing. + """ + if location is None: + return () + return ( + ("x-source-file", location.file), + ("x-source-line", str(location.line)), + ("x-source-function", location.function), + ) + app_identity diff --git a/coordinode/coordinode/client.py b/coordinode/coordinode/client.py index 77ad413..87a4108 100644 --- a/coordinode/coordinode/client.py +++ b/coordinode/coordinode/client.py @@ -14,6 +14,7 @@ import grpc import grpc.aio +from coordinode import _source from coordinode._types import ( PyValue, dict_to_props, @@ -135,6 +136,24 @@ def _make_channel(host: str, port: int, tls: bool) -> grpc.Channel: return grpc.insecure_channel(target) +async def _execute_cypher( + stub: Any, + req: Any, + timeout: float, + metadata: tuple[tuple[str, str], ...], +) -> Any: + """Send one ExecuteCypher, carrying *metadata* only when there is some. + + The argument is omitted rather than passed empty so that a client with + source tracking off makes exactly the call it made before the feature + existed. An opt-in feature should be invisible until it is opted into, + down to the shape of the call. + """ + if metadata: + return await stub.ExecuteCypher(req, timeout=timeout, metadata=metadata) + return await stub.ExecuteCypher(req, timeout=timeout) + + def _make_async_channel(host: str, port: int, tls: bool) -> grpc.aio.Channel: target = f"{host}:{port}" if tls: @@ -504,6 +523,8 @@ async def cypher( self, query: str, params: dict[str, PyValue] | None = None, + *, + _source_location: _source.SourceLocation | None = None, ) -> list[dict[str, Any]]: """Run one statement inside this transaction and return its rows. @@ -536,13 +557,21 @@ async def cypher( parameters=dict_to_props(params or {}), transaction_id=self._id, ) + # Read while this method's caller is still on the stack, which is what + # the helper's frame arithmetic counts on. + metadata = self._client._source_metadata(_source_location) # Transition BEFORE the await, mirroring commit(): a concurrent # commit slipping in while this statement is in flight could land # without the statement's write, and the statement's late "unknown # transaction" failure would then overwrite the real outcome. self._state = "executing" try: - resp = await self._client._cypher_stub.ExecuteCypher(req, timeout=self._client._timeout) + resp = await _execute_cypher( + self._client._cypher_stub, + req, + self._client._timeout, + metadata, + ) except grpc.RpcError: # Always attempt the cleanup, without classifying the failure. # Whether the server processed the statement decides only whether @@ -808,6 +837,15 @@ class AsyncCoordinodeClient: # Also accepts separate host and port: async with AsyncCoordinodeClient("localhost", port=7080) as client: ... + + Pass ``debug_source_tracking=True`` to send the file, line and function + each query was written at. The server's advisor groups its statistics by + query shape, so this is what lets it point at the line that wrote the slow + one instead of only the text, which a dozen call sites usually share. + ``app_name`` and ``app_version`` ride along with it, naming the service the + query came from when several share a database. It is off by default and + costs nothing while off, so turn it on where you are looking, not + everywhere: what it sends is your source paths. """ def __init__( @@ -817,6 +855,9 @@ def __init__( *, tls: bool = False, timeout: float = 30.0, + debug_source_tracking: bool = False, + app_name: str = "", + app_version: str = "", ) -> None: # Support "host:port" as a single string (common gRPC convention). # _HOST_PORT_RE matches "hostname:port" and "[IPv6]:port" but not bare @@ -838,6 +879,12 @@ def __init__( self._port = port self._tls = tls self._timeout = timeout + # Off by default, and off is free: no frame is read and no metadata is + # built unless somebody asked for the attribution. + self._source_tracking = debug_source_tracking + # Constant for the life of the client, so it is built here rather than + # rebuilt for every query. + self._app_identity = _source.identity(app_name, app_version) self._channel: grpc.aio.Channel | None = None # Detached cleanup tasks spawned by cancellation handlers, referenced # here until done (an unreferenced task can be garbage-collected @@ -860,6 +907,27 @@ async def __aenter__(self) -> AsyncCoordinodeClient: async def __aexit__(self, *_: Any) -> None: await self.close() + def _source_metadata( + self, + location: _source.SourceLocation | None, + ) -> tuple[tuple[str, str], ...]: + """gRPC metadata naming the call site, empty unless tracking is on. + + Call this DIRECTLY from the query method whose caller is to be + attributed: with no location supplied it reads the frame two levels up, + which is this helper, then the query method, then the caller. The + synchronous client supplies one instead, because by the time the + coroutine it wrapped runs, the frame that called it is gone. + """ + if not self._source_tracking: + return () + if location is None: + # Two frames up from here: the query method, then its caller. + location = _source.capture(2) + if location is None or not _source.is_call_site(location): + return () + return _source.to_metadata(location, self._app_identity) + async def connect(self) -> None: # A reconnect must not race an in-flight shutdown: when it resumes, # the finalizer unconditionally clears the channel and raises the @@ -955,6 +1023,7 @@ async def cypher( read_preference: str | None = None, after_index: int | None = None, at_timestamp: int | None = None, + _source_location: _source.SourceLocation | None = None, ) -> list[dict[str, Any]]: """Execute an OpenCypher query. Returns rows as list of dicts. @@ -1022,7 +1091,12 @@ async def cypher( req.write_concern.CopyFrom(_make_write_concern(write_concern)) if read_preference is not None: req.read_preference = _make_read_preference(read_preference) - resp = await self._cypher_stub.ExecuteCypher(req, timeout=self._timeout) + resp = await _execute_cypher( + self._cypher_stub, + req, + self._timeout, + self._source_metadata(_source_location), + ) return _rows_to_dicts(resp) def _reclaim_cancelled_begin(self, begin: asyncio.Task[Any]) -> None: @@ -1822,7 +1896,12 @@ def cypher( ) -> list[dict[str, Any]]: """Run one statement inside this transaction. See :meth:`AsyncTransaction.cypher`.""" try: - return self._client._run(self._inner.cypher(query, params)) # type: ignore[no-any-return] + # Read before the coroutine is handed to the loop, for the reason + # given on _caller_location. + location = self._client._caller_location() + return self._client._run( # type: ignore[no-any-return] + self._inner.cypher(query, params, _source_location=location) + ) except BaseException as exc: # An interruption raised INSIDE the stepping coroutine (Ctrl-C # delivered mid-call) completes the task before _run() can @@ -1870,6 +1949,9 @@ class CoordinodeClient: with CoordinodeClient("localhost:7080") as client: rows = client.cypher("MATCH (n:Person) RETURN n.name LIMIT 5") print(rows) # [{"n.name": "Alice"}, ...] + + Takes ``debug_source_tracking``, ``app_name`` and ``app_version`` as + :class:`AsyncCoordinodeClient` does; see there. """ def __init__( @@ -1879,8 +1961,19 @@ def __init__( *, tls: bool = False, timeout: float = 30.0, + debug_source_tracking: bool = False, + app_name: str = "", + app_version: str = "", ) -> None: - self._async = AsyncCoordinodeClient(host, port, tls=tls, timeout=timeout) + self._async = AsyncCoordinodeClient( + host, + port, + tls=tls, + timeout=timeout, + debug_source_tracking=debug_source_tracking, + app_name=app_name, + app_version=app_version, + ) self._loop = asyncio.new_event_loop() self._connected = False @@ -1901,6 +1994,19 @@ def close(self) -> None: if not self._loop.is_closed(): self._loop.close() + def _caller_location(self) -> _source.SourceLocation | None: + """The call site of the synchronous method that calls this, or ``None``. + + Call it DIRECTLY from that method: the frame two levels up is this + helper, then the method, then the caller to attribute. It exists + because the synchronous API hands a coroutine to the loop, and the + coroutine's own view of who called it is the loop by then. + """ + if not self._async._source_tracking: + return None + # Two frames up from here: the synchronous method, then its caller. + return _source.capture(2) + def _run(self, coro: Any) -> Any: if self._loop.is_closed(): raise RuntimeError("CoordinodeClient has been closed and cannot be reused") @@ -1949,6 +2055,10 @@ def cypher( read_preference=read_preference, after_index=after_index, at_timestamp=at_timestamp, + # Read here rather than inside the coroutine: the coroutine + # body runs from the event loop, by which time the frame that + # called this method has already returned. + _source_location=self._caller_location(), ) ) diff --git a/tests/unit/test_source_tracking.py b/tests/unit/test_source_tracking.py new file mode 100644 index 0000000..6098f80 --- /dev/null +++ b/tests/unit/test_source_tracking.py @@ -0,0 +1,325 @@ +"""Unit tests for query source tracking. + +The feature's whole value is that the location it reports is the line a person +wrote the query on, so these tests assert the line rather than the presence of +a header: a test that only checks "some file was sent" passes just as happily +when the file is a frame inside asyncio. + +They also pin the shape of the wire contract, which is shared with the Rust +driver and read by the server's advisor. A renamed key here is silently +ignored on the server, which is exactly the kind of break no runtime error +would report. +""" + +import asyncio +from unittest.mock import AsyncMock + +import grpc +import pytest + +from coordinode._proto.coordinode.v1.query import cypher_pb2 +from coordinode._source import SourceLocation, identity, is_call_site, to_metadata +from coordinode.client import AsyncCoordinodeClient, CoordinodeClient + + +def _execute_response(): + return cypher_pb2.ExecuteCypherResponse(columns=[], rows=[]) + + +def _stub(): + return type( + "FakeCypherStub", + (), + { + "BeginTransaction": AsyncMock(return_value=cypher_pb2.BeginTransactionResponse(transaction_id=42)), + "ExecuteCypher": AsyncMock(return_value=_execute_response()), + "CommitTransaction": AsyncMock(return_value=cypher_pb2.CommitTransactionResponse(applied_index=7)), + "RollbackTransaction": AsyncMock(return_value=cypher_pb2.RollbackTransactionResponse()), + }, + )() + + +def _async_client(**options): + client = AsyncCoordinodeClient("localhost:0", **options) + client._cypher_stub = _stub() + return client + + +def _sync_client(**options): + client = CoordinodeClient("localhost:0", **options) + client._async._cypher_stub = _stub() + client._connected = True + return client + + +def _sent_metadata(stub_method): + """The metadata of the one call made, as a dict. + + An absent argument reads as no metadata, which is what the client sends + when there is none: the keyword is omitted rather than passed empty, so a + query with tracking off makes the same call it made before the feature + existed. + """ + assert stub_method.await_count == 1, f"expected one call, got {stub_method.await_count}" + return dict(stub_method.await_args.kwargs.get("metadata", ())) + + +# ── Off by default ─────────────────────────────────────────────────────────── + + +class TestOffByDefault: + """Tracking sends nothing until it is asked for, and asks the interpreter + for nothing either: the frame is never read, so the cost of the feature + stays with the people who turned it on.""" + + def test_no_metadata_without_the_flag(self): + async def _inner() -> None: + client = _async_client() + await client.cypher("RETURN 1") + call = client._cypher_stub.ExecuteCypher.await_args + # Not merely empty: the keyword is absent, so the default path + # makes the call it made before this feature existed. Test + # doubles written against that signature keep working. + assert "metadata" not in call.kwargs + + asyncio.run(_inner()) + + def test_no_frame_is_read_without_the_flag(self, monkeypatch): + """The flag gates the frame read itself, not just the sending. + + Reading a frame is cheap but not free, and the contract for the + default path is that it does no work at all. A capture that runs + anyway and is then discarded would still pass the test above. + """ + import coordinode.client as client_module + + def _forbidden(*_args, **_kwargs): + raise AssertionError("the caller's frame was read with tracking off") + + # Patched on the client's reference to the module, not on `sys`: + # replacing `sys._getframe` itself would also replace it for the + # logging module, which calls it to name the line a log record came + # from, and the interpreter would take the rest of the suite down + # with it. + monkeypatch.setattr(client_module._source, "capture", _forbidden) + + async def _inner() -> None: + client = _async_client() + await client.cypher("RETURN 1") + + asyncio.run(_inner()) + + +# ── The reported location ──────────────────────────────────────────────────── + + +class TestReportedLocation: + """The line reported is the caller's own, on every path that reaches + ExecuteCypher.""" + + def test_async_query_reports_the_awaiting_line(self): + async def _inner() -> None: + client = _async_client(debug_source_tracking=True) + await client.cypher("RETURN 1") + expected_line = _inner.__code__.co_firstlineno + 2 + + md = _sent_metadata(client._cypher_stub.ExecuteCypher) + assert md["x-source-file"] == __file__ + assert md["x-source-line"] == str(expected_line) + assert md["x-source-function"].endswith("test_async_query_reports_the_awaiting_line.._inner") + + asyncio.run(_inner()) + + def test_sync_query_reports_the_calling_line(self): + """The synchronous client hands a coroutine to its event loop, so by + the time the query runs, this frame has returned. The location has to + be read on the way in or it is gone.""" + client = _sync_client(debug_source_tracking=True) + client.cypher("RETURN 1") + expected_line = self.test_sync_query_reports_the_calling_line.__code__.co_firstlineno + 5 + + md = _sent_metadata(client._async._cypher_stub.ExecuteCypher) + assert md["x-source-file"] == __file__ + assert md["x-source-line"] == str(expected_line) + assert md["x-source-function"].endswith("test_sync_query_reports_the_calling_line") + + def test_transaction_statement_reports_its_own_line(self): + async def _inner() -> None: + client = _async_client(debug_source_tracking=True) + tx = await client.begin_transaction() + await tx.cypher("CREATE (:A)") + expected_line = _inner.__code__.co_firstlineno + 3 + + md = _sent_metadata(client._cypher_stub.ExecuteCypher) + assert md["x-source-line"] == str(expected_line) + + asyncio.run(_inner()) + + def test_sync_transaction_statement_reports_its_own_line(self): + client = _sync_client(debug_source_tracking=True) + with client.transaction() as tx: + tx.cypher("CREATE (:A)") + expected_line = self.test_sync_transaction_statement_reports_its_own_line.__code__.co_firstlineno + 3 + + md = _sent_metadata(client._async._cypher_stub.ExecuteCypher) + assert md["x-source-line"] == str(expected_line) + + +# ── Application identity ───────────────────────────────────────────────────── + + +class TestApplicationIdentity: + """Name and version are optional, and an empty one is not sent: the server + reads a missing key and an empty value the same way, so a header carrying + nothing is only bytes.""" + + def test_name_and_version_ride_with_the_location(self): + async def _inner() -> None: + client = _async_client( + debug_source_tracking=True, + app_name="feed-service", + app_version="2.1.0", + ) + await client.cypher("RETURN 1") + + md = _sent_metadata(client._cypher_stub.ExecuteCypher) + assert md["x-source-app"] == "feed-service" + assert md["x-source-version"] == "2.1.0" + + asyncio.run(_inner()) + + def test_unset_identity_sends_no_key(self): + async def _inner() -> None: + client = _async_client(debug_source_tracking=True, app_name="feed-service") + await client.cypher("RETURN 1") + + md = _sent_metadata(client._cypher_stub.ExecuteCypher) + assert md["x-source-app"] == "feed-service" + assert "x-source-version" not in md + + asyncio.run(_inner()) + + def test_identity_alone_sends_nothing(self): + """Without the flag there is no location, and the server discards a + source context whose file is missing — so name and version alone would + be headers the server throws away.""" + + async def _inner() -> None: + client = _async_client(app_name="feed-service", app_version="2.1.0") + await client.cypher("RETURN 1") + assert _sent_metadata(client._cypher_stub.ExecuteCypher) == {} + + asyncio.run(_inner()) + + +# ── Frames that are not call sites ─────────────────────────────────────────── + + +class TestNonCallSites: + """A frame belonging to this package or to asyncio is what remains when + the frame that wrote the query is already gone. Reporting it would collect + unrelated queries under one event-loop line in the advisor, so nothing is + sent instead.""" + + def test_asyncio_frame_is_rejected(self): + assert not is_call_site(SourceLocation(file=asyncio.__file__, line=1, function="run")) + + def test_sdk_frame_is_rejected(self): + import coordinode.client as client_module + + assert not is_call_site(SourceLocation(file=client_module.__file__, line=1, function="cypher")) + + def test_user_frame_is_accepted(self): + assert is_call_site(SourceLocation(file=__file__, line=1, function="test")) + + def test_neighbour_of_the_package_is_not_swallowed(self): + """The exclusion is by directory, so a path that merely starts with + the same characters — a sibling named `coordinode-extra` beside + `coordinode` — must stay a call site.""" + import os + + import coordinode.client as client_module + + sdk_dir = os.path.dirname(os.path.abspath(client_module.__file__)) + neighbour = f"{sdk_dir}-extra{os.sep}app.py" + assert is_call_site(SourceLocation(file=neighbour, line=1, function="handler")) + + def test_query_started_as_a_task_sends_nothing(self): + """A query handed to create_task runs after the frame that created it + has returned, so the location is genuinely unavailable. Sending the + event-loop frame instead would be a wrong answer, not a partial one.""" + + async def _inner() -> None: + client = _async_client(debug_source_tracking=True) + await asyncio.create_task(client.cypher("RETURN 1")) + assert _sent_metadata(client._cypher_stub.ExecuteCypher) == {} + + asyncio.run(_inner()) + + +# ── Wire contract ──────────────────────────────────────────────────────────── + + +class TestWireContract: + """The keys are shared with the Rust driver and read by the server. A + rename here is not an error anywhere: the server just stops finding the + context, so the names are pinned in a test.""" + + def test_metadata_keys(self): + md = dict( + to_metadata( + SourceLocation(file="app/feed.py", line=47, function="Feed.render"), + identity("feed-service", "2.1.0"), + ) + ) + assert md == { + "x-source-file": "app/feed.py", + "x-source-line": "47", + "x-source-function": "Feed.render", + "x-source-app": "feed-service", + "x-source-version": "2.1.0", + } + + def test_no_location_means_no_metadata(self): + assert to_metadata(None, identity("feed-service", "2.1.0")) == () + + def test_identity_omits_what_was_not_given(self): + assert identity("", "") == () + assert identity("feed-service", "") == (("x-source-app", "feed-service"),) + assert identity("", "2.1.0") == (("x-source-version", "2.1.0"),) + + +# ── Failure paths ──────────────────────────────────────────────────────────── + + +class TestFailurePaths: + """Tracking is a debugging aid and must never be the reason a query + fails.""" + + def test_a_failing_query_still_reports_its_error(self): + class _Rejected(grpc.RpcError): + def code(self): + return grpc.StatusCode.INVALID_ARGUMENT + + async def _inner() -> None: + client = _async_client(debug_source_tracking=True) + client._cypher_stub.ExecuteCypher = AsyncMock(side_effect=_Rejected()) + with pytest.raises(grpc.RpcError): + await client.cypher("RETURN 1") + + asyncio.run(_inner()) + + def test_a_missing_frame_is_not_an_error(self, monkeypatch): + """An interpreter with no Python-level frames, or a stack shorter than + the walk, yields no location. The query goes out unattributed rather + than failing over a debugging aid.""" + import coordinode.client as client_module + + monkeypatch.setattr(client_module._source, "capture", lambda _levels: None) + + async def _inner() -> None: + client = _async_client(debug_source_tracking=True) + await client.cypher("RETURN 1") + assert _sent_metadata(client._cypher_stub.ExecuteCypher) == {} + + asyncio.run(_inner()) From 50ce45a0f97a7b6b9e91dcaf47bf406271c6b595 Mon Sep 17 00:00:00 2001 From: Dmitry Prudnikov Date: Wed, 2 Sep 2026 00:31:57 +0300 Subject: [PATCH 02/10] fix(client): keep the call site through every way of scheduling a query MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Reading the location inside the coroutine only worked for a query that was awaited where it was written. Everything that runs a coroutine as a task starts the body long after the calling frame has returned, so create_task, gather, wait_for, shield and TaskGroup all reported an event-loop frame, which the call-site filter then discarded: concurrent queries, the ones most worth attributing, silently sent nothing. The query methods now read the location when they are CALLED, which is the one moment all of those share, and are plain methods returning a coroutine so there is such a moment at all. Two ways the aid could take a query down with it, both now closed. A path or function name outside ASCII went into a metadata key with no -bin suffix, which carries an HTTP/2 header value and must be ASCII; gRPC enforces that on the client, so the call failed before it was sent and the query never reached the server. A checkout under a non-ASCII path is enough to trigger it, and Python allows such a function name too. The values are escaped now, not dropped: an escaped path still names the file and still groups with itself, and dropping the file key alone would make the server discard the whole context anyway. The frame read caught only ValueError, the failure a stack shorter than the walk gives. An application whose audit hook refuses the sys._getframe event raises whatever it likes instead, and that escaped before the request was built. To a caller all of these are one thing — no location — and none is worth failing a query over. Carries a regression test per fix: the five scheduling wrappers, a location whose file, function and application name are all non-ASCII, and a frame read that raises. All were seen failing first. --- README.md | 12 ++-- coordinode/coordinode/_source.py | 36 +++++++--- coordinode/coordinode/client.py | 87 +++++++++++++++++------ tests/unit/test_source_tracking.py | 106 +++++++++++++++++++++++++++-- 4 files changed, 199 insertions(+), 42 deletions(-) diff --git a/README.md b/README.md index fc3df95..feddf7e 100644 --- a/README.md +++ b/README.md @@ -161,12 +161,12 @@ It is off by default and free while off: no frame is read and the request goes out exactly as it would have. Turn it on where you are looking rather than everywhere, because what it sends is the paths of your source files. -One case reports nothing rather than guessing. A query handed to -`asyncio.create_task` runs after the frame that created it has returned, so its -call site is genuinely gone by then; the alternative would be reporting an -event-loop line, which the advisor would fill with the queries of every -unrelated task that took the same path. `await client.cypher(...)` and the -synchronous client both report normally. +The location is read when you call the method, not when the query runs, so +concurrency does not lose it: `create_task`, `gather`, `wait_for`, `shield` and +`TaskGroup` all start the coroutine long after the calling frame has returned, +and all of them still report the line you wrote. A path or function name +outside ASCII is escaped rather than sent raw, because gRPC refuses a +non-ASCII header and would fail the query rather than the attribution. ## LangChain — GraphRAG Pipeline diff --git a/coordinode/coordinode/_source.py b/coordinode/coordinode/_source.py index 8a88acd..597b06b 100644 --- a/coordinode/coordinode/_source.py +++ b/coordinode/coordinode/_source.py @@ -80,8 +80,12 @@ def capture(levels_up: int) -> SourceLocation | None: answer a question about one of them. ``None`` comes back when the frame cannot be had — an interpreter with no - Python-level frame support, or a stack shorter than the walk. Tracking - goes quiet rather than failing a query over a debugging aid. + Python-level frame support, a stack shorter than the walk, or a hardened + application whose audit hook refuses the ``sys._getframe`` event and + raises whatever it likes in place of an answer. Every one of those is the + same thing to a caller: no location. None of them is worth failing a + query over, which is why the read is caught broadly rather than by the + ValueError a short stack happens to give. """ getframe = getattr(sys, "_getframe", None) if getframe is None: @@ -89,8 +93,7 @@ def capture(levels_up: int) -> SourceLocation | None: try: # +1 for this frame, which the caller counts from rather than into. frame: FrameType = getframe(levels_up + 1) - except ValueError: - # Asked for more stack than exists. + except Exception: return None code = frame.f_code return SourceLocation( @@ -102,6 +105,22 @@ def capture(levels_up: int) -> SourceLocation | None: ) +def _ascii(value: str) -> str: + """*value* with anything gRPC would refuse escaped out. + + A metadata key without the ``-bin`` suffix carries an HTTP/2 header value, + which must be ASCII, and gRPC enforces it on the client: a value with a + non-ASCII character fails the call before it is sent. Both of these can + hold one — a checkout under a non-ASCII path, and a function named in one, + which Python allows — so a debugging aid would become the reason every + query fails. + + Escaped rather than dropped, because an escaped path still names the file + and still groups with itself in the advisor, which is the whole job. + """ + return value.encode("ascii", "backslashreplace").decode("ascii") + + def identity(app_name: str, app_version: str) -> tuple[tuple[str, str], ...]: """The metadata naming the application, built once per client. @@ -114,9 +133,9 @@ def identity(app_name: str, app_version: str) -> tuple[tuple[str, str], ...]: """ pairs = [] if app_name: - pairs.append(("x-source-app", app_name)) + pairs.append(("x-source-app", _ascii(app_name))) if app_version: - pairs.append(("x-source-version", app_version)) + pairs.append(("x-source-version", _ascii(app_version))) return tuple(pairs) @@ -134,7 +153,8 @@ def to_metadata( if location is None: return () return ( - ("x-source-file", location.file), + ("x-source-file", _ascii(location.file)), + # The line is an integer, so its decimal form is ASCII already. ("x-source-line", str(location.line)), - ("x-source-function", location.function), + ("x-source-function", _ascii(location.function)), ) + app_identity diff --git a/coordinode/coordinode/client.py b/coordinode/coordinode/client.py index 87a4108..c4ddb13 100644 --- a/coordinode/coordinode/client.py +++ b/coordinode/coordinode/client.py @@ -7,7 +7,7 @@ import asyncio import logging import re -from collections.abc import AsyncIterator, Iterator, Sequence +from collections.abc import AsyncIterator, Coroutine, Iterator, Sequence from contextlib import asynccontextmanager, contextmanager, suppress from typing import Any @@ -519,13 +519,13 @@ async def _cleanup_preserving_outcome(self) -> None: if _caller_is_being_cancelled(): raise - async def cypher( + def cypher( self, query: str, params: dict[str, PyValue] | None = None, *, _source_location: _source.SourceLocation | None = None, - ) -> list[dict[str, Any]]: + ) -> Coroutine[Any, Any, list[dict[str, Any]]]: """Run one statement inside this transaction and return its rows. The write is buffered rather than applied, so it is visible to later @@ -542,7 +542,21 @@ async def cypher( writes are discarded and the handle is consumed. The failure propagates as-is, and any later use of this object raises instead of reporting the server's "unknown transaction id". + + A plain method returning a coroutine, for the reason given on + :meth:`AsyncCoordinodeClient.cypher`: the call site is read here, while + the caller's frame is still on the stack. """ + return self._cypher(query, params, source_location=_source_location or self._client._call_site()) + + async def _cypher( + self, + query: str, + params: dict[str, PyValue] | None = None, + *, + source_location: _source.SourceLocation | None = None, + ) -> list[dict[str, Any]]: + """Body of :meth:`cypher`; see there.""" from coordinode._proto.coordinode.v1.query.cypher_pb2 import ( # type: ignore[import] ExecuteCypherRequest, ) @@ -557,9 +571,7 @@ async def cypher( parameters=dict_to_props(params or {}), transaction_id=self._id, ) - # Read while this method's caller is still on the stack, which is what - # the helper's frame arithmetic counts on. - metadata = self._client._source_metadata(_source_location) + metadata = self._client._source_metadata(source_location) # Transition BEFORE the await, mirroring commit(): a concurrent # commit slipping in while this statement is in flight could land # without the statement's write, and the statement's late "unknown @@ -907,23 +919,22 @@ async def __aenter__(self) -> AsyncCoordinodeClient: async def __aexit__(self, *_: Any) -> None: await self.close() + def _call_site(self) -> _source.SourceLocation | None: + """Where the public query method that calls this was called from. + + Call it DIRECTLY from that method: two frames up from here is the + method, then the caller to attribute. Nothing is read while tracking + is off. + """ + if not self._source_tracking: + return None + return _source.capture(2) + def _source_metadata( self, location: _source.SourceLocation | None, ) -> tuple[tuple[str, str], ...]: - """gRPC metadata naming the call site, empty unless tracking is on. - - Call this DIRECTLY from the query method whose caller is to be - attributed: with no location supplied it reads the frame two levels up, - which is this helper, then the query method, then the caller. The - synchronous client supplies one instead, because by the time the - coroutine it wrapped runs, the frame that called it is gone. - """ - if not self._source_tracking: - return () - if location is None: - # Two frames up from here: the query method, then its caller. - location = _source.capture(2) + """gRPC metadata naming *location*, empty when there is nothing to say.""" if location is None or not _source.is_call_site(location): return () return _source.to_metadata(location, self._app_identity) @@ -1013,7 +1024,7 @@ async def _finalize_close(self) -> None: # backstop for those. self._closing = True - async def cypher( + def cypher( self, query: str, params: dict[str, PyValue] | None = None, @@ -1024,7 +1035,7 @@ async def cypher( after_index: int | None = None, at_timestamp: int | None = None, _source_location: _source.SourceLocation | None = None, - ) -> list[dict[str, Any]]: + ) -> Coroutine[Any, Any, list[dict[str, Any]]]: """Execute an OpenCypher query. Returns rows as list of dicts. Consistency parameters (all optional; server defaults apply when omitted): @@ -1050,7 +1061,39 @@ async def cypher( non-zero ``after_index``: waiting for a new write and reading a fixed past are opposite requests, and the pair is rejected. Zero is rejected too: it is how the wire says "no pin", so it cannot also ask for one. + + Awaiting this is the whole of its use; it is a plain method returning a + coroutine only so that source tracking can read the call site HERE, + while the caller's frame is still on the stack. Everything that runs a + coroutine as a task — ``create_task``, ``gather``, ``wait_for``, + ``shield``, a ``TaskGroup`` — starts the body long after that frame has + returned, so a location read inside the body would name the event loop + for all of them. """ + return self._cypher( + query, + params, + read_concern=read_concern, + write_concern=write_concern, + read_preference=read_preference, + after_index=after_index, + at_timestamp=at_timestamp, + source_location=_source_location or self._call_site(), + ) + + async def _cypher( + self, + query: str, + params: dict[str, PyValue] | None = None, + *, + read_concern: str | None = None, + write_concern: str | None = None, + read_preference: str | None = None, + after_index: int | None = None, + at_timestamp: int | None = None, + source_location: _source.SourceLocation | None = None, + ) -> list[dict[str, Any]]: + """Body of :meth:`cypher`; see there.""" from coordinode._proto.coordinode.v1.query.cypher_pb2 import ( # type: ignore[import] ExecuteCypherRequest, ) @@ -1095,7 +1138,7 @@ async def cypher( self._cypher_stub, req, self._timeout, - self._source_metadata(_source_location), + self._source_metadata(source_location), ) return _rows_to_dicts(resp) diff --git a/tests/unit/test_source_tracking.py b/tests/unit/test_source_tracking.py index 6098f80..32ae50c 100644 --- a/tests/unit/test_source_tracking.py +++ b/tests/unit/test_source_tracking.py @@ -244,15 +244,68 @@ def test_neighbour_of_the_package_is_not_swallowed(self): neighbour = f"{sdk_dir}-extra{os.sep}app.py" assert is_call_site(SourceLocation(file=neighbour, line=1, function="handler")) - def test_query_started_as_a_task_sends_nothing(self): - """A query handed to create_task runs after the frame that created it - has returned, so the location is genuinely unavailable. Sending the - event-loop frame instead would be a wrong answer, not a partial one.""" +class TestScheduledQueries: + """Everything that schedules the coroutine as a task runs its body after + the frame that wrote the query has returned. The location is therefore + read when the method is CALLED, which is the only moment every one of + these has in common.""" + + def test_create_task(self): async def _inner() -> None: client = _async_client(debug_source_tracking=True) - await asyncio.create_task(client.cypher("RETURN 1")) - assert _sent_metadata(client._cypher_stub.ExecuteCypher) == {} + task = asyncio.create_task(client.cypher("RETURN 1")) + expected_line = _inner.__code__.co_firstlineno + 2 + await task + + md = _sent_metadata(client._cypher_stub.ExecuteCypher) + assert md["x-source-line"] == str(expected_line) + + asyncio.run(_inner()) + + def test_gather(self): + async def _inner() -> None: + client = _async_client(debug_source_tracking=True) + await asyncio.gather(client.cypher("RETURN 1")) + expected_line = _inner.__code__.co_firstlineno + 2 + + md = _sent_metadata(client._cypher_stub.ExecuteCypher) + assert md["x-source-line"] == str(expected_line) + + asyncio.run(_inner()) + + def test_wait_for(self): + async def _inner() -> None: + client = _async_client(debug_source_tracking=True) + await asyncio.wait_for(client.cypher("RETURN 1"), timeout=5) + expected_line = _inner.__code__.co_firstlineno + 2 + + md = _sent_metadata(client._cypher_stub.ExecuteCypher) + assert md["x-source-line"] == str(expected_line) + + asyncio.run(_inner()) + + def test_task_group(self): + async def _inner() -> None: + client = _async_client(debug_source_tracking=True) + async with asyncio.TaskGroup() as tg: + tg.create_task(client.cypher("RETURN 1")) + expected_line = _inner.__code__.co_firstlineno + 3 + + md = _sent_metadata(client._cypher_stub.ExecuteCypher) + assert md["x-source-line"] == str(expected_line) + + asyncio.run(_inner()) + + def test_transaction_statement_as_a_task(self): + async def _inner() -> None: + client = _async_client(debug_source_tracking=True) + tx = await client.begin_transaction() + await asyncio.create_task(tx.cypher("CREATE (:A)")) + expected_line = _inner.__code__.co_firstlineno + 3 + + md = _sent_metadata(client._cypher_stub.ExecuteCypher) + assert md["x-source-line"] == str(expected_line) asyncio.run(_inner()) @@ -283,6 +336,27 @@ def test_metadata_keys(self): def test_no_location_means_no_metadata(self): assert to_metadata(None, identity("feed-service", "2.1.0")) == () + def test_values_are_ascii(self): + """gRPC rejects a non-ASCII value on a key without the -bin suffix, + and rejects it client-side: the query never reaches the server. A + checkout under a non-ASCII path, or a function named in one — Python + allows both — would otherwise turn tracking from a debugging aid into + the reason every query fails. + """ + md = dict( + to_metadata( + SourceLocation(file="/home/пользователь/app.py", line=47, function="Лента.render"), + identity("сервис", "2.1.0"), + ) + ) + for key, value in md.items(): + value.encode("ascii") # raises if the value would be rejected + # Escaped rather than dropped: the path still identifies the file, so + # the advisor can still group by it and a person can still read it. + assert "app.py" in md["x-source-file"] + assert "render" in md["x-source-function"] + assert md["x-source-line"] == "47" + def test_identity_omits_what_was_not_given(self): assert identity("", "") == () assert identity("feed-service", "") == (("x-source-app", "feed-service"),) @@ -309,6 +383,26 @@ async def _inner() -> None: asyncio.run(_inner()) + def test_a_refused_frame_read_is_not_an_error(self, monkeypatch): + """A hardened application can install an audit hook that refuses the + `sys._getframe` event, and the hook's exception comes out of the frame + read rather than the ValueError a short stack gives. Either way the + location is unavailable, and an unavailable location must not take the + query down with it.""" + import coordinode._source as source_module + + def _refused(_depth): + raise RuntimeError("audit hook refused sys._getframe") + + # Scoped to the single call: this replaces the attribute on the `sys` + # module itself, which the logging machinery also reads to name the + # line a record came from, so the substitution must not outlive the + # one call under test. + with monkeypatch.context() as patched: + patched.setattr(source_module.sys, "_getframe", _refused, raising=False) + location = source_module.capture(1) + assert location is None + def test_a_missing_frame_is_not_an_error(self, monkeypatch): """An interpreter with no Python-level frames, or a stack shorter than the walk, yields no location. The query goes out unattributed rather From 2dd8be75e7e4f8e503a98797707b7be15f0fc857 Mon Sep 17 00:00:00 2001 From: Dmitry Prudnikov Date: Wed, 2 Sep 2026 00:44:34 +0300 Subject: [PATCH 03/10] fix(client): escape everything a header cannot carry, not only non-ASCII MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The previous escape covered the wrong range. gRPC refuses a control character in a metadata value exactly as it refuses a non-ASCII one, and 0x7f with them, but backslashreplace leaves all of those untouched: a newline, a tab or a NUL went out raw and failed the call before it was sent. So enabling source tracking could still be the reason every query fails, which is what the escape existed to prevent. The likeliest way in is mundane rather than exotic. An application name read from a file arrives with the newline that ended it, and a POSIX path may legally contain one. The bar is now printable ASCII, stated as a predicate rather than delegated to a codec whose range would have to be verified. Values that need nothing — every ordinary path and name — are returned unchanged after two scans, so the common path allocates nothing where it previously built a new string every time. Carries a regression test with a newline, a tab and a NUL across the file, function and application values, seen failing first. The escaped output was also put through grpc to confirm it now reaches the transport, where the raw one was rejected. --- README.md | 7 +++--- coordinode/coordinode/_source.py | 35 ++++++++++++++++++++---------- tests/unit/test_source_tracking.py | 18 +++++++++++++++ 3 files changed, 45 insertions(+), 15 deletions(-) diff --git a/README.md b/README.md index feddf7e..301001e 100644 --- a/README.md +++ b/README.md @@ -164,9 +164,10 @@ everywhere, because what it sends is the paths of your source files. The location is read when you call the method, not when the query runs, so concurrency does not lose it: `create_task`, `gather`, `wait_for`, `shield` and `TaskGroup` all start the coroutine long after the calling frame has returned, -and all of them still report the line you wrote. A path or function name -outside ASCII is escaped rather than sent raw, because gRPC refuses a -non-ASCII header and would fail the query rather than the attribution. +and all of them still report the line you wrote. Anything outside printable +ASCII is escaped rather than sent raw — a non-ASCII path, but a newline or a +tab just as much, since gRPC refuses those in a header too and would fail the +query rather than the attribution. ## LangChain — GraphRAG Pipeline diff --git a/coordinode/coordinode/_source.py b/coordinode/coordinode/_source.py index 597b06b..45e63ca 100644 --- a/coordinode/coordinode/_source.py +++ b/coordinode/coordinode/_source.py @@ -105,20 +105,31 @@ def capture(levels_up: int) -> SourceLocation | None: ) -def _ascii(value: str) -> str: - """*value* with anything gRPC would refuse escaped out. +def _header_safe(value: str) -> str: + """*value* with everything gRPC would refuse escaped out. A metadata key without the ``-bin`` suffix carries an HTTP/2 header value, - which must be ASCII, and gRPC enforces it on the client: a value with a - non-ASCII character fails the call before it is sent. Both of these can - hold one — a checkout under a non-ASCII path, and a function named in one, - which Python allows — so a debugging aid would become the reason every - query fails. + and gRPC enforces the permitted range on the client: a value outside + printable ASCII fails the call before it is sent, so an unescaped one + would make a debugging aid the reason every query fails. + + The bar is printable ASCII, not ASCII. A newline, a tab, a NUL and 0x7f + are each refused the same way a non-ASCII character is, and the likeliest + source of one is mundane: an application name read from a file arrives + with the newline that ended it. A POSIX path may legally contain one too. Escaped rather than dropped, because an escaped path still names the file and still groups with itself in the advisor, which is the whole job. """ - return value.encode("ascii", "backslashreplace").decode("ascii") + # Both checks are single scans in C, and together they say exactly + # "every character is in 0x20..0x7e": isascii rules out the rest of + # Unicode, isprintable rules out the control characters and 0x7f while + # counting the space as printable. + if value.isascii() and value.isprintable(): + return value + return "".join( + ch if " " <= ch <= "~" else f"\\x{ord(ch):02x}" if ord(ch) < 0x100 else f"\\u{ord(ch):04x}" for ch in value + ) def identity(app_name: str, app_version: str) -> tuple[tuple[str, str], ...]: @@ -133,9 +144,9 @@ def identity(app_name: str, app_version: str) -> tuple[tuple[str, str], ...]: """ pairs = [] if app_name: - pairs.append(("x-source-app", _ascii(app_name))) + pairs.append(("x-source-app", _header_safe(app_name))) if app_version: - pairs.append(("x-source-version", _ascii(app_version))) + pairs.append(("x-source-version", _header_safe(app_version))) return tuple(pairs) @@ -153,8 +164,8 @@ def to_metadata( if location is None: return () return ( - ("x-source-file", _ascii(location.file)), + ("x-source-file", _header_safe(location.file)), # The line is an integer, so its decimal form is ASCII already. ("x-source-line", str(location.line)), - ("x-source-function", _ascii(location.function)), + ("x-source-function", _header_safe(location.function)), ) + app_identity diff --git a/tests/unit/test_source_tracking.py b/tests/unit/test_source_tracking.py index 32ae50c..271583b 100644 --- a/tests/unit/test_source_tracking.py +++ b/tests/unit/test_source_tracking.py @@ -357,6 +357,24 @@ def test_values_are_ascii(self): assert "render" in md["x-source-function"] assert md["x-source-line"] == "47" + def test_values_are_printable(self): + """ASCII alone is not the bar: gRPC refuses a control character in a + header value just as it refuses a non-ASCII one, and 0x7f with them. + An application name read from a file arrives with the trailing + newline, and a POSIX path may legally contain one, so this is the + likelier of the two ways to break every query.""" + md = dict( + to_metadata( + SourceLocation(file="/app/feed\n.py", line=47, function="render\x00"), + identity("feed-service\n", "2.1.0\t"), + ) + ) + for value in md.values(): + assert all(" " <= ch <= "~" for ch in value), repr(value) + assert "feed" in md["x-source-file"] + assert "render" in md["x-source-function"] + assert "feed-service" in md["x-source-app"] + def test_identity_omits_what_was_not_given(self): assert identity("", "") == () assert identity("feed-service", "") == (("x-source-app", "feed-service"),) From 67222fb5a7ca79eb6ef279021e367443932f45e1 Mon Sep 17 00:00:00 2001 From: Dmitry Prudnikov Date: Wed, 2 Sep 2026 01:06:27 +0300 Subject: [PATCH 04/10] fix(client): keep the coroutine contract and stop the escaping merging call sites Two ways the previous round went wrong. Making the query methods plain methods returning a coroutine took the call site back from every scheduling wrapper, but it also changed the answer to a question callers ask: iscoroutinefunction said False. Code that dispatches on it broke, most consequentially create_autospec, which then built a synchronous double whose result a test cannot await. That happened with tracking off, so it fell on people not using the feature at all. The methods are now marked as standing in for the coroutine functions they replace, both ways, since the lever differs by version. The escaping was not injective, and for this feature that is not a lost detail but a wrong answer: the advisor groups by what it receives, so two call sites arriving as one string are reported as one place in the code, with the queries of one attributed to the other. A name holding a real newline came out as a name holding the four characters that spell its escape, and a character outside the basic plane came out as a character inside it followed by a digit. The escape character now escapes itself and each form is padded to a fixed width. Carries a regression test per defect, each seen failing first: both introspection predicates and an autospecced double that gets awaited, a newline against its literal spelling, and an astral character against a BMP one followed by a digit. The encoding was also brute-forced over an alphabet of the characters that trip it, three deep: no collisions, nothing outside the permitted range. --- coordinode/coordinode/_source.py | 38 +++++++++++++++---- coordinode/coordinode/client.py | 26 +++++++++++++ tests/unit/test_source_tracking.py | 61 ++++++++++++++++++++++++++++++ 3 files changed, 117 insertions(+), 8 deletions(-) diff --git a/coordinode/coordinode/_source.py b/coordinode/coordinode/_source.py index 45e63ca..fb70521 100644 --- a/coordinode/coordinode/_source.py +++ b/coordinode/coordinode/_source.py @@ -118,18 +118,40 @@ def _header_safe(value: str) -> str: source of one is mundane: an application name read from a file arrives with the newline that ended it. A POSIX path may legally contain one too. + The encoding is injective, which matters more than it looks: two call + sites arriving as one string would not merely lose detail, they would be + reported as one place in the code, with the queries of one attributed to + the other. Two things buy that. The escape character escapes itself, so a + name holding a real newline differs from one holding the characters that + spell its escape. And each form is padded to a fixed width, so a character + outside the basic plane cannot spell the same thing as one inside it + followed by a digit. + Escaped rather than dropped, because an escaped path still names the file and still groups with itself in the advisor, which is the whole job. """ - # Both checks are single scans in C, and together they say exactly - # "every character is in 0x20..0x7e": isascii rules out the rest of - # Unicode, isprintable rules out the control characters and 0x7f while - # counting the space as printable. - if value.isascii() and value.isprintable(): + # Three single scans in C, and together they say "every character is in + # 0x20..0x7e, and none of them is the escape character": isascii rules out + # the rest of Unicode, isprintable rules out the control characters and + # 0x7f while counting the space as printable, and a value with no + # backslash cannot collide with an escaped one. + if value.isascii() and value.isprintable() and "\\" not in value: return value - return "".join( - ch if " " <= ch <= "~" else f"\\x{ord(ch):02x}" if ord(ch) < 0x100 else f"\\u{ord(ch):04x}" for ch in value - ) + return "".join(_escape(ch) for ch in value) + + +def _escape(ch: str) -> str: + """One character as itself, or as a fixed-width escape. See _header_safe.""" + if ch == "\\": + return "\\\\" + if " " <= ch <= "~": + return ch + point = ord(ch) + if point < 0x100: + return f"\\x{point:02x}" + if point < 0x10000: + return f"\\u{point:04x}" + return f"\\U{point:08x}" def identity(app_name: str, app_version: str) -> tuple[tuple[str, str], ...]: diff --git a/coordinode/coordinode/client.py b/coordinode/coordinode/client.py index c4ddb13..a7333d3 100644 --- a/coordinode/coordinode/client.py +++ b/coordinode/coordinode/client.py @@ -5,6 +5,7 @@ from __future__ import annotations import asyncio +import inspect import logging import re from collections.abc import AsyncIterator, Coroutine, Iterator, Sequence @@ -136,6 +137,29 @@ def _make_channel(host: str, port: int, tls: bool) -> grpc.Channel: return grpc.insecure_channel(target) +def _stands_in_for_a_coroutine_function(fn: Any) -> Any: + """Mark *fn* so introspection sees the coroutine function it replaces. + + The query methods are plain methods returning a coroutine, so that the + call site can be read while the caller is still on the stack. A caller + still writes ``await client.cypher(...)``, but code that ASKS instead of + awaiting would get the wrong answer: `unittest.mock.create_autospec` reads + the predicate to decide whether to build an async double, and a + synchronous double hands back a plain value where the test awaits one. + + Marked two ways because the levers differ by version: from 3.12 there is a + supported one, and before it the answer comes from an attribute that + ``asyncio.iscoroutinefunction`` (and therefore mock) reads. + """ + mark = getattr(inspect, "markcoroutinefunction", None) + if mark is not None: + return mark(fn) + marker = getattr(asyncio.coroutines, "_is_coroutine", None) + if marker is not None: + fn._is_coroutine = marker + return fn + + async def _execute_cypher( stub: Any, req: Any, @@ -519,6 +543,7 @@ async def _cleanup_preserving_outcome(self) -> None: if _caller_is_being_cancelled(): raise + @_stands_in_for_a_coroutine_function def cypher( self, query: str, @@ -1024,6 +1049,7 @@ async def _finalize_close(self) -> None: # backstop for those. self._closing = True + @_stands_in_for_a_coroutine_function def cypher( self, query: str, diff --git a/tests/unit/test_source_tracking.py b/tests/unit/test_source_tracking.py index 271583b..51194a3 100644 --- a/tests/unit/test_source_tracking.py +++ b/tests/unit/test_source_tracking.py @@ -245,6 +245,42 @@ def test_neighbour_of_the_package_is_not_swallowed(self): assert is_call_site(SourceLocation(file=neighbour, line=1, function="handler")) +class TestIntrospection: + """The query methods stand in for the coroutine functions they used to be, + and everything that asks must still get that answer. The capture had to + move out of the body, but a caller still writes `await client.cypher(...)`, + and code that dispatches on the predicate — `unittest.mock.create_autospec` + most consequentially — must keep building async doubles. This holds with + tracking off, which is the default, so the cost of getting it wrong falls + on people not using the feature at all.""" + + def test_client_cypher_is_a_coroutine_function(self): + import inspect + + assert inspect.iscoroutinefunction(AsyncCoordinodeClient.cypher) + assert asyncio.iscoroutinefunction(AsyncCoordinodeClient.cypher) + + def test_transaction_cypher_is_a_coroutine_function(self): + import inspect + + from coordinode.client import AsyncTransaction + + assert inspect.iscoroutinefunction(AsyncTransaction.cypher) + assert asyncio.iscoroutinefunction(AsyncTransaction.cypher) + + def test_autospec_builds_an_async_double(self): + """The concrete breakage: an autospecced client whose `cypher` came out + synchronous returns a plain value where the caller awaits.""" + from unittest.mock import create_autospec + + async def _inner() -> None: + double = create_autospec(AsyncCoordinodeClient, instance=True) + double.cypher.return_value = [{"n": 1}] + assert await double.cypher("RETURN 1") == [{"n": 1}] + + asyncio.run(_inner()) + + class TestScheduledQueries: """Everything that schedules the coroutine as a task runs its body after the frame that wrote the query has returned. The location is therefore @@ -357,6 +393,31 @@ def test_values_are_ascii(self): assert "render" in md["x-source-function"] assert md["x-source-line"] == "47" + def test_escaping_does_not_merge_distinct_files(self): + """Two different files must not arrive as the same metadata. + + The advisor groups by what it receives, so a collision does not lose + information, it invents it: the queries of one call site are reported + under another. A file whose name holds a real newline and a file whose + name holds the four characters that spell the escape are exactly such + a pair, which is why the escape character escapes itself. + """ + with_newline = to_metadata(SourceLocation(file="/app/a\n.py", line=1, function="f"), ()) + with_literal = to_metadata(SourceLocation(file="/app/a\\x0a.py", line=1, function="f"), ()) + assert dict(with_newline)["x-source-file"] != dict(with_literal)["x-source-file"] + + def test_escaping_is_fixed_width_per_form(self): + """The other way two names could arrive as one string. + + An escape whose length varied with the code point would let a + character outside the basic plane spell the same thing as a character + inside it followed by a digit. Each form is therefore padded to its + own fixed width, as Python's own escapes are. + """ + astral = to_metadata(SourceLocation(file="\U0001f600", line=1, function="f"), ()) + bmp_then_digit = to_metadata(SourceLocation(file="ὠ0", line=1, function="f"), ()) + assert dict(astral)["x-source-file"] != dict(bmp_then_digit)["x-source-file"] + def test_values_are_printable(self): """ASCII alone is not the bar: gRPC refuses a control character in a header value just as it refuses a non-ASCII one, and 0x7f with them. From 821e27e29b9de5eef584893138d682d1a350da37 Mon Sep 17 00:00:00 2001 From: Dmitry Prudnikov Date: Wed, 2 Sep 2026 02:24:21 +0300 Subject: [PATCH 05/10] fix(client): read the call site at lookup, keeping the methods coroutines Marking a plain method as standing in for a coroutine function only worked from 3.12, where the marker is honoured. On 3.11 it reaches asyncio.iscoroutinefunction and therefore mock, but inspect.iscoroutinefunction has no such lever and answered False, which the tests caught there. The methods are coroutine functions again, and the location is read at the moment that serves both halves: attribute lookup. It happens in the caller's own frame, and it happens for every way of running the query, since create_task, gather, wait_for, shield and a TaskGroup all begin with `client.cypher(...)`. What the caller gets is a partial over the coroutine function, which inspect looks through, so the predicates are right on every supported version rather than on the newest ones. The signature is declared without self, because a descriptor that is not a plain function is not recognised as a method by create_autospec, which then leaves self in and binds the first real argument to it. With tracking off the lookup returns the ordinary bound method: no partial, no frame read, nothing. Verified on 3.11 as well as 3.12, since the version was the whole point: both predicates on the class, on a transaction and on a tracking instance, and an autospecced double that gets awaited. --- coordinode/coordinode/client.py | 154 +++++++++++++++----------------- 1 file changed, 73 insertions(+), 81 deletions(-) diff --git a/coordinode/coordinode/client.py b/coordinode/coordinode/client.py index a7333d3..ba3e3a9 100644 --- a/coordinode/coordinode/client.py +++ b/coordinode/coordinode/client.py @@ -5,10 +5,11 @@ from __future__ import annotations import asyncio +import functools import inspect import logging import re -from collections.abc import AsyncIterator, Coroutine, Iterator, Sequence +from collections.abc import AsyncIterator, Iterator, Sequence from contextlib import asynccontextmanager, contextmanager, suppress from typing import Any @@ -137,27 +138,52 @@ def _make_channel(host: str, port: int, tls: bool) -> grpc.Channel: return grpc.insecure_channel(target) -def _stands_in_for_a_coroutine_function(fn: Any) -> Any: - """Mark *fn* so introspection sees the coroutine function it replaces. - - The query methods are plain methods returning a coroutine, so that the - call site can be read while the caller is still on the stack. A caller - still writes ``await client.cypher(...)``, but code that ASKS instead of - awaiting would get the wrong answer: `unittest.mock.create_autospec` reads - the predicate to decide whether to build an async double, and a - synchronous double hands back a plain value where the test awaits one. - - Marked two ways because the levers differ by version: from 3.12 there is a - supported one, and before it the answer comes from an attribute that - ``asyncio.iscoroutinefunction`` (and therefore mock) reads. +class _tracks_its_call_site: # noqa: N801 — reads as a decorator at the use site + """Read the caller's location when the method is looked up, not when its + coroutine runs. + + Both moments matter and they are different. A query handed to + ``create_task``, ``gather``, ``wait_for``, ``shield`` or a ``TaskGroup`` + starts its body long after the frame that wrote it has returned, so a + location read inside the body names an event-loop frame for all of them. + Attribute lookup, on the other hand, happens in the caller's own frame, + every time and for every one of those, since they all begin with + ``client.cypher(...)``. + + The method it wraps stays a coroutine function, and what a caller gets is + a partial over it. That keeps both halves of the contract: the location is + read at the right moment, and code that ASKS whether this is a coroutine + function still gets yes — `inspect.iscoroutinefunction` looks through a + partial, and `unittest.mock.create_autospec`, which reads the class, finds + the coroutine function itself. A double that came out synchronous would + hand back a plain value where the test awaits one. + + With tracking off there is no partial and no frame read: the lookup + returns the ordinary bound method. """ - mark = getattr(inspect, "markcoroutinefunction", None) - if mark is not None: - return mark(fn) - marker = getattr(asyncio.coroutines, "_is_coroutine", None) - if marker is not None: - fn._is_coroutine = marker - return fn + + def __init__(self, fn: Any) -> None: + self._fn = fn + functools.update_wrapper(self, fn) + # Drop `self` from the advertised signature. A descriptor that is not + # a plain function is not recognised as a method by + # `unittest.mock.create_autospec`, which then leaves `self` in and + # binds the first real argument to it, so an autospecced call reports + # the wrong argument missing. Saying the signature outright is what + # the caller sees anyway, since the method is always reached through + # an instance. + parameters = list(inspect.signature(fn).parameters.values())[1:] + fn.__signature__ = inspect.Signature(parameters) + + def __get__(self, obj: Any, objtype: Any = None) -> Any: + if obj is None: + return self._fn + if not obj._source_tracking_enabled(): + return self._fn.__get__(obj, objtype) + # One frame up from here is whoever wrote `obj.cypher`. A location + # bound as a keyword stays a default: the synchronous client passes + # its own, read at ITS boundary, and that one wins. + return functools.partial(self._fn, obj, _source_location=_source.capture(1)) async def _execute_cypher( @@ -397,6 +423,14 @@ def is_open(self) -> bool: """True while the transaction can still take statements and be committed.""" return self._state == "open" + def _source_tracking_enabled(self) -> bool: + """Whether a statement's lookup should read its call site. + + A transaction has no setting of its own: it is the client's, since the + connection is what carries the tracking. + """ + return self._client._source_tracking + def _require_open(self, action: str) -> None: if self._state == "open": return @@ -543,14 +577,14 @@ async def _cleanup_preserving_outcome(self) -> None: if _caller_is_being_cancelled(): raise - @_stands_in_for_a_coroutine_function - def cypher( + @_tracks_its_call_site + async def cypher( self, query: str, params: dict[str, PyValue] | None = None, *, _source_location: _source.SourceLocation | None = None, - ) -> Coroutine[Any, Any, list[dict[str, Any]]]: + ) -> list[dict[str, Any]]: """Run one statement inside this transaction and return its rows. The write is buffered rather than applied, so it is visible to later @@ -568,20 +602,9 @@ def cypher( as-is, and any later use of this object raises instead of reporting the server's "unknown transaction id". - A plain method returning a coroutine, for the reason given on - :meth:`AsyncCoordinodeClient.cypher`: the call site is read here, while - the caller's frame is still on the stack. + The call site is read at lookup, for the reason given on + :meth:`AsyncCoordinodeClient.cypher`. """ - return self._cypher(query, params, source_location=_source_location or self._client._call_site()) - - async def _cypher( - self, - query: str, - params: dict[str, PyValue] | None = None, - *, - source_location: _source.SourceLocation | None = None, - ) -> list[dict[str, Any]]: - """Body of :meth:`cypher`; see there.""" from coordinode._proto.coordinode.v1.query.cypher_pb2 import ( # type: ignore[import] ExecuteCypherRequest, ) @@ -596,7 +619,7 @@ async def _cypher( parameters=dict_to_props(params or {}), transaction_id=self._id, ) - metadata = self._client._source_metadata(source_location) + metadata = self._client._source_metadata(_source_location) # Transition BEFORE the await, mirroring commit(): a concurrent # commit slipping in while this statement is in flight could land # without the statement's write, and the statement's late "unknown @@ -944,16 +967,13 @@ async def __aenter__(self) -> AsyncCoordinodeClient: async def __aexit__(self, *_: Any) -> None: await self.close() - def _call_site(self) -> _source.SourceLocation | None: - """Where the public query method that calls this was called from. + def _source_tracking_enabled(self) -> bool: + """Whether to read the call site when a query method is looked up. - Call it DIRECTLY from that method: two frames up from here is the - method, then the caller to attribute. Nothing is read while tracking - is off. + Read by the lookup itself, so it has to be answerable without + touching anything else: with tracking off, nothing at all happens. """ - if not self._source_tracking: - return None - return _source.capture(2) + return self._source_tracking def _source_metadata( self, @@ -1049,8 +1069,8 @@ async def _finalize_close(self) -> None: # backstop for those. self._closing = True - @_stands_in_for_a_coroutine_function - def cypher( + @_tracks_its_call_site + async def cypher( self, query: str, params: dict[str, PyValue] | None = None, @@ -1061,7 +1081,7 @@ def cypher( after_index: int | None = None, at_timestamp: int | None = None, _source_location: _source.SourceLocation | None = None, - ) -> Coroutine[Any, Any, list[dict[str, Any]]]: + ) -> list[dict[str, Any]]: """Execute an OpenCypher query. Returns rows as list of dicts. Consistency parameters (all optional; server defaults apply when omitted): @@ -1088,38 +1108,10 @@ def cypher( opposite requests, and the pair is rejected. Zero is rejected too: it is how the wire says "no pin", so it cannot also ask for one. - Awaiting this is the whole of its use; it is a plain method returning a - coroutine only so that source tracking can read the call site HERE, - while the caller's frame is still on the stack. Everything that runs a - coroutine as a task — ``create_task``, ``gather``, ``wait_for``, - ``shield``, a ``TaskGroup`` — starts the body long after that frame has - returned, so a location read inside the body would name the event loop - for all of them. + The call site is read when this method is looked up rather than when + its body runs, because everything that starts a coroutine as a task + runs the body after the calling frame has returned. """ - return self._cypher( - query, - params, - read_concern=read_concern, - write_concern=write_concern, - read_preference=read_preference, - after_index=after_index, - at_timestamp=at_timestamp, - source_location=_source_location or self._call_site(), - ) - - async def _cypher( - self, - query: str, - params: dict[str, PyValue] | None = None, - *, - read_concern: str | None = None, - write_concern: str | None = None, - read_preference: str | None = None, - after_index: int | None = None, - at_timestamp: int | None = None, - source_location: _source.SourceLocation | None = None, - ) -> list[dict[str, Any]]: - """Body of :meth:`cypher`; see there.""" from coordinode._proto.coordinode.v1.query.cypher_pb2 import ( # type: ignore[import] ExecuteCypherRequest, ) @@ -1164,7 +1156,7 @@ async def _cypher( self._cypher_stub, req, self._timeout, - self._source_metadata(source_location), + self._source_metadata(_source_location), ) return _rows_to_dicts(resp) From e961277604e3ef41028bcdc2ea84cd65b6ad1b95 Mon Sep 17 00:00:00 2001 From: Dmitry Prudnikov Date: Wed, 2 Sep 2026 11:06:52 +0300 Subject: [PATCH 06/10] fix(client): keep query in the signature of a bound query method Rewriting the underlying function's signature to drop `self` was read again every time Python bound the method, taking the first remaining parameter off with it: `inspect.signature(client.cypher)` reported `params` first and no `query` at all, with tracking off as well as on. Anything inspecting a bound callable to validate arguments, inject dependencies or generate a wrapper would build that interface. The self-less signature exists for `create_autospec`, which reaches the method through the class, so it now lives on a separate object returned by class-level access only. The function every binding is derived from keeps its true signature. Carries a regression test on both bindings and on a transaction statement. Part of #78 --- coordinode/coordinode/client.py | 39 ++++++++++++++++++++++-------- tests/unit/test_source_tracking.py | 35 +++++++++++++++++++++++++++ 2 files changed, 64 insertions(+), 10 deletions(-) diff --git a/coordinode/coordinode/client.py b/coordinode/coordinode/client.py index ba3e3a9..6081c42 100644 --- a/coordinode/coordinode/client.py +++ b/coordinode/coordinode/client.py @@ -138,6 +138,33 @@ def _make_channel(host: str, port: int, tls: bool) -> grpc.Channel: return grpc.insecure_channel(target) +def _without_self(fn: Any) -> Any: + """*fn* again, advertising the parameters a caller actually passes. + + Only for reaching the method through the CLASS, which is how + `unittest.mock.create_autospec` reads it. A descriptor that is not a plain + function is not recognised as a method there, so `self` is left in and the + first real argument binds to it — an autospecced call then reports the + wrong argument missing. + + It has to be a separate object. Rewriting the original function's own + signature would be read again every time Python binds the method, taking + the first REMAINING parameter off with it: `client.cypher` would advertise + itself as taking no `query`, and anything inspecting a bound callable to + validate arguments, inject dependencies or generate a wrapper would build + that interface. The original therefore keeps its true signature, which is + the one every binding is derived from. + """ + + @functools.wraps(fn) + async def unbound(*args: Any, **kwargs: Any) -> Any: + return await fn(*args, **kwargs) + + parameters = list(inspect.signature(fn).parameters.values())[1:] + unbound.__signature__ = inspect.Signature(parameters) # type: ignore[attr-defined] + return unbound + + class _tracks_its_call_site: # noqa: N801 — reads as a decorator at the use site """Read the caller's location when the method is looked up, not when its coroutine runs. @@ -165,19 +192,11 @@ class _tracks_its_call_site: # noqa: N801 — reads as a decorator at the use s def __init__(self, fn: Any) -> None: self._fn = fn functools.update_wrapper(self, fn) - # Drop `self` from the advertised signature. A descriptor that is not - # a plain function is not recognised as a method by - # `unittest.mock.create_autospec`, which then leaves `self` in and - # binds the first real argument to it, so an autospecced call reports - # the wrong argument missing. Saying the signature outright is what - # the caller sees anyway, since the method is always reached through - # an instance. - parameters = list(inspect.signature(fn).parameters.values())[1:] - fn.__signature__ = inspect.Signature(parameters) + self._unbound = _without_self(fn) def __get__(self, obj: Any, objtype: Any = None) -> Any: if obj is None: - return self._fn + return self._unbound if not obj._source_tracking_enabled(): return self._fn.__get__(obj, objtype) # One frame up from here is whoever wrote `obj.cypher`. A location diff --git a/tests/unit/test_source_tracking.py b/tests/unit/test_source_tracking.py index 51194a3..20f21e1 100644 --- a/tests/unit/test_source_tracking.py +++ b/tests/unit/test_source_tracking.py @@ -496,3 +496,38 @@ async def _inner() -> None: assert _sent_metadata(client._cypher_stub.ExecuteCypher) == {} asyncio.run(_inner()) + + +class TestBoundSignatures: + """Looking a query method up must not cost it a parameter. + + The descriptor tells `unittest.mock.create_autospec` what the method's + parameters are, and the obvious way to do that — rewriting the underlying + function's signature without `self` — is read again every time Python + binds the method, taking `query` off with it. Frameworks that inspect a + bound callable to validate arguments, inject dependencies or generate a + wrapper would then build the wrong interface, and would do it with + tracking off as well, which is the default. + """ + + def test_bound_query_signature_keeps_query(self): + import inspect + + client = _async_client() + assert "query" in inspect.signature(client.cypher).parameters + + def test_tracked_bound_query_signature_keeps_query(self): + import inspect + + client = _async_client(debug_source_tracking=True) + assert "query" in inspect.signature(client.cypher).parameters + + def test_bound_transaction_signature_keeps_query(self): + import inspect + + async def _inner() -> None: + client = _async_client(debug_source_tracking=True) + tx = await client.begin_transaction() + assert "query" in inspect.signature(tx.cypher).parameters + + asyncio.run(_inner()) From 074808a08bc8d604371c916db44b9ff28926cf37 Mon Sep 17 00:00:00 2001 From: Dmitry Prudnikov Date: Wed, 2 Sep 2026 11:08:23 +0300 Subject: [PATCH 07/10] fix(client): attribute a saved query method to the line that calls it Reading the call site when the method was looked up filed every query of a bound method kept for later under the one line that stored it: `run_query = client.cypher` followed by calls from a dozen places reported the assignment. Dependency injection and callback-style code hold on to a bound method routinely, so the advisor would merge exactly the call sites the feature exists to tell apart. The read moves into the call, which is the moment every path shares: `client.cypher(...)` evaluates in the caller's own frame whatever is then done with the coroutine, so scheduling it as a task still reports the caller and not the event loop. The coroutine contract holds because the wrapper derives from functools.partial, which the predicates unwrap. Carries a regression test for a saved method, and for two of its calls staying two distinct locations. Part of #78 --- coordinode/coordinode/client.py | 70 +++++++++++++++++------------- tests/unit/test_source_tracking.py | 40 +++++++++++++++++ 2 files changed, 81 insertions(+), 29 deletions(-) diff --git a/coordinode/coordinode/client.py b/coordinode/coordinode/client.py index 6081c42..1b8dc33 100644 --- a/coordinode/coordinode/client.py +++ b/coordinode/coordinode/client.py @@ -165,28 +165,43 @@ async def unbound(*args: Any, **kwargs: Any) -> Any: return unbound +class _bound_query(functools.partial): # noqa: N801 — an implementation detail, named like one + """A query method bound to its client, reading the call site when CALLED. + + The moment matters, and two nearby ones are wrong. Inside the coroutine's + body is too late: a query handed to ``create_task``, ``gather``, + ``wait_for``, ``shield`` or a ``TaskGroup`` starts its body long after the + frame that wrote it has returned, so the location would name an + event-loop frame for all of them. Attribute lookup is too early: a bound + method kept for later — which dependency injection and callback-style code + do routinely — is looked up once at the wiring and called from everywhere + afterwards, so every one of those queries would be filed under the line + that stored it. The call is the moment they all share: ``client.cypher(…)`` + evaluates in the caller's own frame whatever is then done with the + coroutine. + + Deriving from ``functools.partial`` is what keeps the coroutine contract. + Code that ASKS whether this is a coroutine function still gets yes, since + the predicates unwrap partials to the function underneath, and a double + that came out synchronous would hand back a plain value where the caller + awaits one. + """ + + def __call__(self, /, *args: Any, **kwargs: Any) -> Any: + # One frame up is whoever wrote the call. A location passed in stays: + # the synchronous client and the helper methods read theirs at THEIR + # boundary, which is the caller's frame rather than this package's. + kwargs.setdefault("_source_location", _source.capture(1)) + return super().__call__(*args, **kwargs) + + class _tracks_its_call_site: # noqa: N801 — reads as a decorator at the use site - """Read the caller's location when the method is looked up, not when its - coroutine runs. - - Both moments matter and they are different. A query handed to - ``create_task``, ``gather``, ``wait_for``, ``shield`` or a ``TaskGroup`` - starts its body long after the frame that wrote it has returned, so a - location read inside the body names an event-loop frame for all of them. - Attribute lookup, on the other hand, happens in the caller's own frame, - every time and for every one of those, since they all begin with - ``client.cypher(...)``. - - The method it wraps stays a coroutine function, and what a caller gets is - a partial over it. That keeps both halves of the contract: the location is - read at the right moment, and code that ASKS whether this is a coroutine - function still gets yes — `inspect.iscoroutinefunction` looks through a - partial, and `unittest.mock.create_autospec`, which reads the class, finds - the coroutine function itself. A double that came out synchronous would - hand back a plain value where the test awaits one. - - With tracking off there is no partial and no frame read: the lookup - returns the ordinary bound method. + """Make a query method report where each of its calls was written. + + With tracking on, a lookup hands back a :class:`_bound_query`, which reads + the caller when it is called. With tracking off there is no wrapper and no + frame read: the lookup returns the ordinary bound method, and the query + makes the call it made before this feature existed. """ def __init__(self, fn: Any) -> None: @@ -199,10 +214,7 @@ def __get__(self, obj: Any, objtype: Any = None) -> Any: return self._unbound if not obj._source_tracking_enabled(): return self._fn.__get__(obj, objtype) - # One frame up from here is whoever wrote `obj.cypher`. A location - # bound as a keyword stays a default: the synchronous client passes - # its own, read at ITS boundary, and that one wins. - return functools.partial(self._fn, obj, _source_location=_source.capture(1)) + return _bound_query(self._fn, obj) async def _execute_cypher( @@ -621,7 +633,7 @@ async def cypher( as-is, and any later use of this object raises instead of reporting the server's "unknown transaction id". - The call site is read at lookup, for the reason given on + The call site is read when the call is made, for the reason given on :meth:`AsyncCoordinodeClient.cypher`. """ from coordinode._proto.coordinode.v1.query.cypher_pb2 import ( # type: ignore[import] @@ -1127,9 +1139,9 @@ async def cypher( opposite requests, and the pair is rejected. Zero is rejected too: it is how the wire says "no pin", so it cannot also ask for one. - The call site is read when this method is looked up rather than when - its body runs, because everything that starts a coroutine as a task - runs the body after the calling frame has returned. + The call site is read when this method is called rather than when its + body runs, because everything that starts a coroutine as a task runs + the body after the calling frame has returned. """ from coordinode._proto.coordinode.v1.query.cypher_pb2 import ( # type: ignore[import] ExecuteCypherRequest, diff --git a/tests/unit/test_source_tracking.py b/tests/unit/test_source_tracking.py index 20f21e1..bea117d 100644 --- a/tests/unit/test_source_tracking.py +++ b/tests/unit/test_source_tracking.py @@ -531,3 +531,43 @@ async def _inner() -> None: assert "query" in inspect.signature(tx.cypher).parameters asyncio.run(_inner()) + + +class TestSavedBoundMethods: + """A query method kept and called later reports where it was CALLED. + + Dependency injection and callback-style code routinely hold on to a bound + method, so a location read when the method is looked up would file every + later query under the one line that did the wiring — exactly the merging + of unrelated call sites the feature exists to undo. + """ + + def test_saved_bound_method_reports_the_invoking_line(self): + async def _inner() -> None: + client = _async_client(debug_source_tracking=True) + run_query = client.cypher + await run_query("RETURN 1") + expected_line = _inner.__code__.co_firstlineno + 3 + + md = _sent_metadata(client._cypher_stub.ExecuteCypher) + assert md["x-source-line"] == str(expected_line) + + asyncio.run(_inner()) + + def test_two_calls_of_one_saved_method_report_two_lines(self): + """The consequence that matters: distinct call sites stay distinct.""" + + async def _inner() -> None: + client = _async_client(debug_source_tracking=True) + run_query = client.cypher + await run_query("RETURN 1") + await run_query("RETURN 2") + first_line = _inner.__code__.co_firstlineno + 3 + + lines = [ + dict(call.kwargs.get("metadata", ()))["x-source-line"] + for call in client._cypher_stub.ExecuteCypher.await_args_list + ] + assert lines == [str(first_line), str(first_line + 1)] + + asyncio.run(_inner()) From c249d6c877c0f21be4f4d812c18fe333eee248f6 Mon Sep 17 00:00:00 2001 From: Dmitry Prudnikov Date: Wed, 2 Sep 2026 11:10:14 +0300 Subject: [PATCH 08/10] fix(client): attribute the text-index helpers to their own callers create_text_index and drop_text_index reach the server through an internal cypher call, so the location read there was a frame inside this package: not a call site, discarded, and two public query paths left silently outside the per-query attribution the client advertises. Both helpers now read their own caller and pass it down to the statement, on the asynchronous and the synchronous client alike. Carries a regression test per helper on each client, and one holding the default path quiet. Part of #78 --- coordinode/coordinode/client.py | 33 ++++++++++++++--- tests/unit/test_source_tracking.py | 59 ++++++++++++++++++++++++++++++ 2 files changed, 87 insertions(+), 5 deletions(-) diff --git a/coordinode/coordinode/client.py b/coordinode/coordinode/client.py index 1b8dc33..c3a3650 100644 --- a/coordinode/coordinode/client.py +++ b/coordinode/coordinode/client.py @@ -1765,6 +1765,7 @@ async def create_edge_type( et = await self._schema_stub.CreateEdgeType(req, timeout=self._timeout) return EdgeTypeInfo(et) + @_tracks_its_call_site async def create_text_index( self, name: str, @@ -1772,6 +1773,7 @@ async def create_text_index( properties: str | list[str] | tuple[str, ...], *, language: str = "", + _source_location: _source.SourceLocation | None = None, ) -> TextIndexInfo: """Create a full-text (BM25) index on one or more node properties. @@ -1815,7 +1817,10 @@ async def create_text_index( props_expr = ", ".join(prop_list) lang_clause = f" DEFAULT LANGUAGE {language}" if language else "" cypher = f"CREATE TEXT INDEX {name} ON :{label}({props_expr}){lang_clause}" - rows = await self.cypher(cypher) + # Passed on rather than left to the statement below to read: that + # lookup happens here, inside the package, which is not a call site + # and would leave this public query path unattributed. + rows = await self.cypher(cypher, _source_location=_source_location) if rows: return TextIndexInfo(rows[0]) effective_language = language or "english" @@ -1823,7 +1828,13 @@ async def create_text_index( {"index": name, "label": label, "properties": ", ".join(prop_list), "default_language": effective_language} ) - async def drop_text_index(self, name: str) -> None: + @_tracks_its_call_site + async def drop_text_index( + self, + name: str, + *, + _source_location: _source.SourceLocation | None = None, + ) -> None: """Drop a full-text index by name. Args: @@ -1837,7 +1848,9 @@ async def drop_text_index(self, name: str) -> None: await client.drop_text_index("article_body") """ _validate_cypher_identifier(name, "name") - await self.cypher(f"DROP TEXT INDEX {name}") + # The caller's location, passed on for the reason given in + # create_text_index. + await self.cypher(f"DROP TEXT INDEX {name}", _source_location=_source_location) async def traverse( self, @@ -2311,11 +2324,21 @@ def create_text_index( language: str = "", ) -> TextIndexInfo: """Create a full-text (BM25) index on one or more node properties.""" - return self._run(self._async.create_text_index(name, label, properties, language=language)) + return self._run( + self._async.create_text_index( + name, + label, + properties, + language=language, + # Read here rather than deeper in, for the reason given on + # _caller_location. + _source_location=self._caller_location(), + ) + ) def drop_text_index(self, name: str) -> None: """Drop a full-text index by name.""" - return self._run(self._async.drop_text_index(name)) + return self._run(self._async.drop_text_index(name, _source_location=self._caller_location())) def traverse( self, diff --git a/tests/unit/test_source_tracking.py b/tests/unit/test_source_tracking.py index bea117d..eb024fc 100644 --- a/tests/unit/test_source_tracking.py +++ b/tests/unit/test_source_tracking.py @@ -571,3 +571,62 @@ async def _inner() -> None: assert lines == [str(first_line), str(first_line + 1)] asyncio.run(_inner()) + + +class TestHelperQueries: + """The public helpers that reach the server through `cypher` are attributed + to the line that called the HELPER. + + Their internal `self.cypher(...)` sits in this package, so a location read + there is not a call site and is dropped — leaving two public query paths + silently outside the per-query attribution the client advertises. + """ + + def test_create_text_index_reports_the_calling_line(self): + async def _inner() -> None: + client = _async_client(debug_source_tracking=True) + await client.create_text_index("article_body", "Article", "body") + expected_line = _inner.__code__.co_firstlineno + 2 + + md = _sent_metadata(client._cypher_stub.ExecuteCypher) + assert md["x-source-file"] == __file__ + assert md["x-source-line"] == str(expected_line) + + asyncio.run(_inner()) + + def test_drop_text_index_reports_the_calling_line(self): + async def _inner() -> None: + client = _async_client(debug_source_tracking=True) + await client.drop_text_index("article_body") + expected_line = _inner.__code__.co_firstlineno + 2 + + md = _sent_metadata(client._cypher_stub.ExecuteCypher) + assert md["x-source-line"] == str(expected_line) + + asyncio.run(_inner()) + + def test_sync_create_text_index_reports_the_calling_line(self): + client = _sync_client(debug_source_tracking=True) + client.create_text_index("article_body", "Article", "body") + expected_line = self.test_sync_create_text_index_reports_the_calling_line.__code__.co_firstlineno + 2 + + md = _sent_metadata(client._async._cypher_stub.ExecuteCypher) + assert md["x-source-file"] == __file__ + assert md["x-source-line"] == str(expected_line) + + def test_sync_drop_text_index_reports_the_calling_line(self): + client = _sync_client(debug_source_tracking=True) + client.drop_text_index("article_body") + expected_line = self.test_sync_drop_text_index_reports_the_calling_line.__code__.co_firstlineno + 2 + + md = _sent_metadata(client._async._cypher_stub.ExecuteCypher) + assert md["x-source-line"] == str(expected_line) + + def test_helper_query_still_unattributed_without_the_flag(self): + async def _inner() -> None: + client = _async_client() + await client.drop_text_index("article_body") + call = client._cypher_stub.ExecuteCypher.await_args + assert "metadata" not in call.kwargs + + asyncio.run(_inner()) From 4db84b3c8e336434be77d19baf8c34c44d6f4e1f Mon Sep 17 00:00:00 2001 From: Dmitry Prudnikov Date: Wed, 2 Sep 2026 11:22:17 +0300 Subject: [PATCH 09/10] fix(client): survive an audit hook that refuses the frame's attributes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Reading `f_code` raises its own `object.__getattr__` audit event, apart from the `sys._getframe` one, so a hardened application permitting the first and rejecting the second had the exception come out of the read — past the guard, and into a query that then failed rather than going out unattributed. A debugging aid must never be why a query fails, whichever of the two events the hook objects to. The extraction moves inside the guard that already covers getting the frame, which is the same nothing to a caller either way. Carries a regression test per refused attribute, and one for the query that still goes out. Part of #78 --- coordinode/coordinode/_source.py | 32 +++++++----- tests/unit/test_source_tracking.py | 83 ++++++++++++++++++++++++++++++ 2 files changed, 102 insertions(+), 13 deletions(-) diff --git a/coordinode/coordinode/_source.py b/coordinode/coordinode/_source.py index fb70521..bf7b780 100644 --- a/coordinode/coordinode/_source.py +++ b/coordinode/coordinode/_source.py @@ -81,11 +81,12 @@ def capture(levels_up: int) -> SourceLocation | None: ``None`` comes back when the frame cannot be had — an interpreter with no Python-level frame support, a stack shorter than the walk, or a hardened - application whose audit hook refuses the ``sys._getframe`` event and - raises whatever it likes in place of an answer. Every one of those is the - same thing to a caller: no location. None of them is worth failing a - query over, which is why the read is caught broadly rather than by the - ValueError a short stack happens to give. + application whose audit hook refuses the ``sys._getframe`` event, or the + ``object.__getattr__`` one that reading the frame's own attributes + raises, and answers with whatever exception it likes. Every one of those + is the same thing to a caller: no location. None of them is worth failing + a query over, which is why the whole read is caught broadly rather than + by the ValueError a short stack happens to give. """ getframe = getattr(sys, "_getframe", None) if getframe is None: @@ -93,16 +94,21 @@ def capture(levels_up: int) -> SourceLocation | None: try: # +1 for this frame, which the caller counts from rather than into. frame: FrameType = getframe(levels_up + 1) + # Reading the frame's attributes is guarded with the read of the frame + # itself, because it is a second thing an audit hook can refuse: + # CPython raises `object.__getattr__` when `f_code` is read, separately + # from the `sys._getframe` event, and a hook may object to either. Both + # leave the same nothing behind, and neither is worth a failed query. + code = frame.f_code + return SourceLocation( + file=code.co_filename, + line=frame.f_lineno, + # Qualified, so a method reads as "Class.method" rather than a bare + # name that says nothing about which class it belongs to. + function=code.co_qualname, + ) except Exception: return None - code = frame.f_code - return SourceLocation( - file=code.co_filename, - line=frame.f_lineno, - # Qualified, so a method reads as "Class.method" rather than a bare - # name that says nothing about which class it belongs to. - function=code.co_qualname, - ) def _header_safe(value: str) -> str: diff --git a/tests/unit/test_source_tracking.py b/tests/unit/test_source_tracking.py index eb024fc..bf7113e 100644 --- a/tests/unit/test_source_tracking.py +++ b/tests/unit/test_source_tracking.py @@ -630,3 +630,86 @@ async def _inner() -> None: assert "metadata" not in call.kwargs asyncio.run(_inner()) + + +class TestUnboundCallForm: + """Reaching the method through the class, with the instance passed in, is + a valid way to call it, and it must still be attributed. + + That form bypasses the binding entirely, so nothing on the way in reads + the caller. A client with tracking on would send the query with no + location at all — quietly, since an unattributed query is exactly what + the feature's own failure paths produce. + """ + + def test_unbound_call_reports_the_awaiting_line(self): + async def _inner() -> None: + client = _async_client(debug_source_tracking=True) + await AsyncCoordinodeClient.cypher(client, "RETURN 1") + expected_line = _inner.__code__.co_firstlineno + 2 + + md = _sent_metadata(client._cypher_stub.ExecuteCypher) + assert md["x-source-file"] == __file__ + assert md["x-source-line"] == str(expected_line) + + asyncio.run(_inner()) + + def test_unbound_call_sends_nothing_without_the_flag(self): + async def _inner() -> None: + client = _async_client() + await AsyncCoordinodeClient.cypher(client, "RETURN 1") + call = client._cypher_stub.ExecuteCypher.await_args + assert "metadata" not in call.kwargs + + asyncio.run(_inner()) + + +class TestRefusedFrameAttributes: + """An audit hook can refuse the frame's attributes as readily as the frame + itself. + + CPython raises a separate `object.__getattr__` event when `f_code` is + read, so a hook that permits `sys._getframe` and rejects that one would + have the exception come out of the read — and, since the query is already + on its way, fail it. A debugging aid must never be the reason a query + fails, whichever of the two events the hook objects to. + """ + + def _frame_refusing(self, attribute): + class _Frame: + def __getattr__(self, name): + if name == attribute: + raise RuntimeError(f"audit hook refused frame.{name}") + raise AssertionError(f"unexpected attribute {name}") + + return _Frame() + + @pytest.mark.parametrize("attribute", ["f_code", "f_lineno"]) + def test_a_refused_frame_attribute_yields_no_location(self, monkeypatch, attribute): + import coordinode._source as source_module + + with monkeypatch.context() as patched: + patched.setattr( + source_module.sys, + "_getframe", + lambda _depth: self._frame_refusing(attribute), + raising=False, + ) + assert source_module.capture(1) is None + + def test_a_refused_frame_attribute_does_not_fail_the_query(self, monkeypatch): + import coordinode._source as source_module + + async def _inner() -> None: + client = _async_client(debug_source_tracking=True) + with monkeypatch.context() as patched: + patched.setattr( + source_module.sys, + "_getframe", + lambda _depth: self._frame_refusing("f_code"), + raising=False, + ) + await client.cypher("RETURN 1") + assert _sent_metadata(client._cypher_stub.ExecuteCypher) == {} + + asyncio.run(_inner()) From 969ef9d284a006a8213be02d62aa7a42347705d0 Mon Sep 17 00:00:00 2001 From: Dmitry Prudnikov Date: Wed, 2 Sep 2026 11:23:09 +0300 Subject: [PATCH 10/10] fix(client): attribute a query called through the class MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The unbound form — passing the instance in, as AsyncCoordinodeClient.cypher(client, "...") — reaches the method through the class and so bypasses the binding entirely. Nothing on the way in read the caller, and a client with tracking on sent that query with no location at all: quiet, since an unattributed query is also what every unreadable frame produces. It now reads its caller as well. From inside the coroutine, which is as early as this form allows and right for the direct await it is written as; scheduling this particular form as a task lands on an event-loop frame and is left unattributed rather than misattributed, as every unreadable location is here. Carries a regression test for the attributed call and for the default path staying silent. Part of #78 --- coordinode/coordinode/client.py | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/coordinode/coordinode/client.py b/coordinode/coordinode/client.py index c3a3650..3e5bc17 100644 --- a/coordinode/coordinode/client.py +++ b/coordinode/coordinode/client.py @@ -154,10 +154,20 @@ def _without_self(fn: Any) -> Any: validate arguments, inject dependencies or generate a wrapper would build that interface. The original therefore keeps its true signature, which is the one every binding is derived from. + + Calling it is a real, if uncommon, way to run a query — + ``AsyncCoordinodeClient.cypher(client, "…")`` passes the instance itself — + so it reads the call site too. It reads it from inside the coroutine, + which is as early as this form allows and right for the direct ``await`` + that is how it is written; a caller who instead schedules THIS form as a + task lands on an event-loop frame and is left unattributed rather than + misattributed, which is what every unreadable location does here. """ @functools.wraps(fn) async def unbound(*args: Any, **kwargs: Any) -> Any: + if args and kwargs.get("_source_location") is None and args[0]._source_tracking_enabled(): + kwargs["_source_location"] = _source.capture(1) return await fn(*args, **kwargs) parameters = list(inspect.signature(fn).parameters.values())[1:]