From 8dda58601ec59d8be8ea9b94d0f16020eab0ed35 Mon Sep 17 00:00:00 2001 From: Taewoon Kim Date: Thu, 12 Nov 2026 10:13:28 +0900 Subject: [PATCH 1/3] arcadedb.aio.AsyncDatabase: await query, command, run_in_transaction and find_nearest from async def code (refs #306) An asyncio facade over the blocking calls of an open Database, on a bounded thread pool of its own (max_workers, default 8). Decided on #306: a transaction belongs to one thread and two awaits of one task can run on two workers, so there is no async with transaction; run_in_transaction(fn) runs the plain function fn, begin to commit and every retry, on one worker. Cancellation is not exposed: cancelling the awaiting task abandons the result, a call that started finishes, a call still waiting for a worker does not run, and close() waits for the calls that were submitted (it does not close the Database). The docstring and the docs page say how AsyncDatabase differs from the engine's AsyncExecutor. tests/test_aio.py pins, on both wheels: rows and errors reach the awaiting task, run_in_transaction commits on one worker and rolls back on an exception, 200 tasks write and read at once without a lost update, the pool runs max_workers calls at once and never more, close() waits and refuses new calls, a cancelled await abandons the result. tests/test_docs_asyncio_page.py runs the blocks of the new page how-to/asyncio.adoc. Items 3 and 4 of #306 still need measurements. Co-Authored-By: Claude Sonnet 5.5 --- docs/modules/ROOT/nav.adoc | 1 + .../ROOT/pages/how-to/async-executor.adoc | 2 +- docs/modules/ROOT/pages/how-to/asyncio.adoc | 221 +++++++++++++ src/arcadedb/aio.py | 171 ++++++++++ tests/test_aio.py | 305 ++++++++++++++++++ tests/test_docs_asyncio_page.py | 60 ++++ 6 files changed, 759 insertions(+), 1 deletion(-) create mode 100644 docs/modules/ROOT/pages/how-to/asyncio.adoc create mode 100644 src/arcadedb/aio.py create mode 100644 tests/test_aio.py create mode 100644 tests/test_docs_asyncio_page.py diff --git a/docs/modules/ROOT/nav.adoc b/docs/modules/ROOT/nav.adoc index c0d9de6..459853c 100644 --- a/docs/modules/ROOT/nav.adoc +++ b/docs/modules/ROOT/nav.adoc @@ -7,6 +7,7 @@ ** xref:how-to/transactions.adoc[Transactions] ** xref:how-to/values-and-records.adoc[Values and records] ** xref:how-to/concurrency.adoc[Concurrency, processes and crashes] +** xref:how-to/asyncio.adoc[asyncio: AsyncDatabase] ** xref:how-to/durability.adoc[Durability settings] ** xref:how-to/async-executor.adoc[The async executor] ** xref:how-to/logs.adoc[Where the logs go] diff --git a/docs/modules/ROOT/pages/how-to/async-executor.adoc b/docs/modules/ROOT/pages/how-to/async-executor.adoc index 23cf062..242acb5 100644 --- a/docs/modules/ROOT/pages/how-to/async-executor.adoc +++ b/docs/modules/ROOT/pages/how-to/async-executor.adoc @@ -3,7 +3,7 @@ // Sources: src/arcadedb/async_executor.py (the JVM build), src/arcadedb/_native/extras.py (NativeAsyncExecutor, the native build), core.py (async_executor), and ArcadeDB 26.11.1 DatabaseAsyncExecutorImpl.java. Pinned by tests/test_async_executor.py, tests/test_async_transaction.py and tests/test_async_executor_behavior.py. -You need this page if you want to hand statements to the engine and carry on, and learn about the result in a callback. The short version: +You need this page if you want to hand statements to the engine and carry on, and learn about the result in a callback. For `async def` code that awaits each call, use `AsyncDatabase` instead: see xref:how-to/asyncio.adoc[asyncio: AsyncDatabase]. The short version: * `executor = db.async_executor()` returns a wrapper of the one executor of the database. Every call returns another wrapper of the same executor, with the same settings. * `executor.command(...)`, `executor.query(...)` and `executor.transaction(...)` queue work and return at once. `executor.wait_completion()` blocks until everything queued before it is stored. diff --git a/docs/modules/ROOT/pages/how-to/asyncio.adoc b/docs/modules/ROOT/pages/how-to/asyncio.adoc new file mode 100644 index 0000000..1107614 --- /dev/null +++ b/docs/modules/ROOT/pages/how-to/asyncio.adoc @@ -0,0 +1,221 @@ += asyncio: AsyncDatabase +:description: Call a Database from async def code with arcadedb.aio.AsyncDatabase: awaited query, command, run_in_transaction and find_nearest on a bounded thread pool, what a transaction, a cancel and close() mean there, and how it differs from the async executor. + +// Sources: src/arcadedb/aio.py. Pinned by tests/test_aio.py, tests/test_docs_asyncio_page.py and tests/test_vector_read_your_writes.py. + +You need this page if your code is written with `async def`, for example an agent loop that stores a few vectors per turn and searches right after, and a database call must not block the event loop. The short version: + +* `adb = AsyncDatabase(db, max_workers=8)` wraps an open `Database`. It owns a pool of `max_workers` threads, and it is a thin layer: the calls and the errors are those of `db`. +* `await adb.query(...)`, `await adb.command(...)` and `await adb.find_nearest(index, ...)` run the blocking call on a worker and return its result. `await adb.run(fn, *args)` does the same for any function. +* A transaction belongs to one thread, and two awaits of one task can run on two workers. So there is no `async with adb.transaction()`: `await adb.run_in_transaction(fn)` runs the plain function `fn`, from begin to commit and through every retry, on one worker. +* Cancelling the task that awaits a call abandons the result. A call that has started on a worker still finishes. A call that still waits for a worker does not run. +* `await adb.close()` refuses new calls and waits for the ones that were submitted. It does not close the `Database`. + +The statements are SQL or openCypher, as on xref:how-to/databases-and-queries.adoc[Databases and queries]. This page covers what the facade adds. What the engine does with several threads is on xref:how-to/concurrency.adoc[Concurrency, processes and crashes]. + +The blocks on this page run one after the other against one empty database, and a test in the repository runs them that way (`tests/test_docs_asyncio_page.py`) on the JVM build and on the native build. Each block that prints something is followed by its output. Every block assumes that `db` is an open database. + +== Open it, write and read + +`AsyncDatabase` is an async context manager, so `close()` runs when the block ends. A write goes in `run_in_transaction`, because a transaction cannot span awaits. A read needs no transaction. `query` returns a list of `dict`, the rows as `ResultSet.to_list()` gives them, read on the worker. + +[source,python] +---- +import asyncio + +from arcadedb.aio import AsyncDatabase + +db.command("sql", "CREATE DOCUMENT TYPE Note") + + +def add_hello(): + db.command("sql", "INSERT INTO Note SET text = ?", "hello") + + +async def main(): + async with AsyncDatabase(db, max_workers=4) as adb: + await adb.run_in_transaction(add_hello) + rows = await adb.query("sql", "SELECT text FROM Note") + print([row["text"] for row in rows]) + + +asyncio.run(main()) +---- + +[source,text] +---- +['hello'] +---- + +Parameters bind as in `db.query`: positional `?` from the extra arguments, or by name from one `dict`. `command` returns a list of `dict` as well, or `None` when the engine returns no result set. An `INSERT`, `UPDATE` or `DELETE` through `command` needs a transaction, so send it through `run_in_transaction`, or switch on `db.set_auto_transaction(True)`, which commits each statement on its own (see xref:how-to/transactions.adoc#auto-transaction[Transactions]). Do not send `BEGIN` or `COMMIT` through `command`: the next call can run on another worker, where that transaction is not open. + +== Many tasks at once + +Start as many tasks as you like. The pool runs at most `max_workers` calls at a time, and the others wait for a worker. Writers that change the same record conflict, and `run_in_transaction` runs the function again after such a conflict: pass `retries` for a contended record, as on xref:how-to/concurrency.adoc#several-writers[Several writers]. + +[source,python] +---- +async def remember(adb, n): + def write(): + db.command("sql", "INSERT INTO Note SET text = ?", f"note {n}") + + await adb.run_in_transaction(write, retries=60) + + +async def main(): + async with AsyncDatabase(db) as adb: + await asyncio.gather(*(remember(adb, n) for n in range(100))) + rows = await adb.query("sql", "SELECT count(*) AS n FROM Note") + print(rows[0]["n"]) + + +asyncio.run(main()) +---- + +[source,text] +---- +101 +---- + +The default of 8 workers is a starting point, not a measured optimum. Measure it on your own workload before you change it. + +== A transaction that fails + +`fn` is a plain function with no arguments. It makes blocking calls on the `Database`, and it contains no `await`. An exception of `fn` rolls the transaction back and is raised in the awaiting task. The next block stores a note inside the function and then fails, and the note is gone. + +[source,python] +---- +def fail(): + db.command("sql", "INSERT INTO Note SET text = 'never stored'") + raise ValueError("not this one") + + +async def main(): + async with AsyncDatabase(db) as adb: + try: + await adb.run_in_transaction(fail) + except ValueError as error: + print("raised:", error) + rows = await adb.query("sql", "SELECT count(*) AS n FROM Note") + print(rows[0]["n"]) + + +asyncio.run(main()) +---- + +[source,text] +---- +raised: not this one +101 +---- + +== Any other call + +`run(fn, *args, **kwargs)` runs any blocking function on a worker and returns its result. Use it for what the facade does not wrap, and for a result shape other than a list of `dict`. The context variables of the calling task are visible to `fn`, as with `asyncio.to_thread`. + +[source,python] +---- +def first_text(): + return db.query("sql", "SELECT text FROM Note ORDER BY text LIMIT 1").first().get("text") + + +async def main(): + async with AsyncDatabase(db) as adb: + print(await adb.run(first_text)) + + +asyncio.run(main()) +---- + +[source,text] +---- +hello +---- + +== Vector search + +`find_nearest(index, query_vector, k=10, ...)` takes the `VectorIndex` that `db.create_vector_index(...)` or `db.schema.get_vector_index(...)` returns, and gives what `index.find_nearest` gives: a list of `(record, distance)`, nearest first. A search finds every vector whose transaction committed before the search started, so an agent can insert in one call and search in the next, from the same task or from another one: see xref:how-to/vectors.adoc#agents-that-write-as-they-go[Agents that write as they go]. + +[source,python] +---- +db.command("sql", "CREATE VERTEX TYPE Memory") +db.command("sql", "CREATE PROPERTY Memory.embedding ARRAY_OF_FLOATS") +with db.transaction(): + for name, vector in [ + ("red", [1.0, 0.0, 0.0]), + ("green", [0.0, 1.0, 0.0]), + ("blue", [0.0, 0.0, 1.0]), + ]: + db.command( + "sql", "INSERT INTO Memory SET name = ?, embedding = ?", name, vector + ) +index = db.create_vector_index("Memory", "embedding", dimensions=3) + + +def add_orange(): + db.command( + "sql", + "INSERT INTO Memory SET name = ?, embedding = ?", + "orange", + [0.9, 0.1, 0.0], + ) + + +async def main(): + async with AsyncDatabase(db) as adb: + await adb.run_in_transaction(add_orange) + hits = await adb.find_nearest(index, [0.9, 0.1, 0.0], k=1) + print(hits[0][0].get("name")) + + +asyncio.run(main()) +---- + +[source,text] +---- +orange +---- + +== What runs where + +Every call runs on a thread of the pool of the facade, never on the event loop, and the pool is its own: the facade does not use the default executor of the loop. The engine call runs without the Python GIL on both builds (JPype releases it for a Java call, `ctypes` for a call into the native library), so the loop keeps running while the engine works. Building the Python rows from the answer, which `query` and `command` do on the worker, holds the GIL. How late the loop wakes up during such calls was measured on xref:how-to/concurrency.adoc#asyncio[Concurrency, processes and crashes] for `asyncio.to_thread`, which is the same mechanism. + +One `Database` object shared by the workers is supported on both builds: see xref:how-to/concurrency.adoc#threads[Threads]. A worker keeps nothing between two calls, as long as every transaction is opened and ended inside one `run_in_transaction`. + +== Cancel and close + +Cancellation is not part of the facade. Cancelling the task that awaits a call, with `task.cancel()` or a timeout, stops the wait and drops the result. What happens to the call depends on where it is: + +* A call that still waits for a worker does not run. +* A call that already runs on a worker is not interrupted, and finishes. A transaction function still commits or rolls back on its own worker, so no transaction is left half open. + +`await adb.close()` sets the facade to closed, so a new call raises `RuntimeError`, and then waits for the calls that were submitted, the queued ones included. The wait does not block the event loop. A second `close()` waits for the same calls. The `Database` stays open: close it yourself. + +== AsyncDatabase or the async executor + +Both names say "async", and they are different tools. + +[cols="2,3,3"] +|=== +| | `AsyncDatabase` (`arcadedb.aio`) | `AsyncExecutor` (`db.async_executor()`) + +| Made by +| `AsyncDatabase(db, max_workers=8)` +| `db.async_executor()`, the one executor of the database + +| Runs on +| Python threads of its own pool, through the blocking API +| The threads of the engine + +| A call +| Awaited: returns its result, or raises in the awaiting task +| Queued: returns at once, and the result arrives in a callback + +| Transactions +| `run_in_transaction(fn)`: one worker from begin to commit, with retries +| Each worker commits after `commit_every` operations and at `wait_completion()` + +| Use it for +| `async def` code: reads, vector search and small transactions +| Work that the engine runs in the background while your code carries on: see xref:how-to/async-executor.adoc[The async executor] +|=== diff --git a/src/arcadedb/aio.py b/src/arcadedb/aio.py new file mode 100644 index 0000000..5ce82b1 --- /dev/null +++ b/src/arcadedb/aio.py @@ -0,0 +1,171 @@ +"""asyncio facade over the blocking calls of an open :class:`~arcadedb.Database` (``arcadedb.aio``). + +An agent loop written with ``async def`` can call the bindings through :class:`AsyncDatabase` instead of wrapping every call in +``asyncio.to_thread`` itself. Each call runs the ordinary blocking call on a worker thread of the facade's own pool, so the event loop +keeps running while the engine works. It is a thin layer: no engine change, the same calls, the same errors. + +Three rules shape it: + +* **A transaction belongs to one thread.** The engine ties a transaction to the thread that began it, and two awaits of one task can + run on two different workers, so there is no ``async with adb.transaction()``. :meth:`AsyncDatabase.run_in_transaction` takes a plain + function and runs it, begin to commit and every retry, on one worker. +* **One pool, bounded.** The facade owns a :class:`concurrent.futures.ThreadPoolExecutor` of ``max_workers`` threads (default 8) and + never takes the loop's default executor. One ``Database`` shared by several threads is supported: see the concurrency page. +* **Cancellation is not exposed.** Cancelling the task that awaits a call abandons the result. A call that already started on a worker + finishes anyway (a transaction function still commits or rolls back on its own thread). A call that still waits for a worker does + not run. :meth:`AsyncDatabase.close` waits for the calls that were submitted. + +``AsyncDatabase`` and :class:`~arcadedb.AsyncExecutor` (``db.async_executor()``) are different things. ``AsyncExecutor`` is the engine's +own asynchronous API: statements are queued on the engine's threads, the call returns at once, the result arrives in a callback, and it +is meant for fire-and-forget bulk writes. ``AsyncDatabase`` is for ``async def`` code: every call is awaited, returns its own result or +raises in the awaiting task, and runs through the normal blocking API on a Python thread pool. +""" + +from __future__ import annotations + +import asyncio +import contextvars +import functools +from concurrent.futures import ThreadPoolExecutor +from typing import Any, Callable, Dict, List, Optional + +__all__ = ["AsyncDatabase"] + +_DEFAULT_MAX_WORKERS = 8 + + +class AsyncDatabase: + """Await the calls of an open ``Database`` from ``asyncio`` code. + + Args: + db: An open :class:`~arcadedb.Database`. The facade does not close it; closing the facade does not either. + max_workers: The number of worker threads, so the most calls that run at once. Default 8. It is a starting point, not a + measured optimum: writers that touch the same records conflict, and ``run_in_transaction`` retries them. + + Use it as ``async with AsyncDatabase(db) as adb:`` or call ``await adb.close()`` yourself. + """ + + def __init__(self, db, max_workers: int = _DEFAULT_MAX_WORKERS): + if ( + isinstance(max_workers, bool) + or not isinstance(max_workers, int) + or max_workers < 1 + ): + raise ValueError(f"max_workers must be an int >= 1, got {max_workers!r}") + self._db = db + self._max_workers = max_workers + self._executor = ThreadPoolExecutor( + max_workers=max_workers, thread_name_prefix="arcadedb-aio" + ) + self._closed = False + + @property + def database(self): + """The wrapped ``Database``.""" + return self._db + + @property + def max_workers(self) -> int: + """The size of the worker pool.""" + return self._max_workers + + @property + def closed(self) -> bool: + """``True`` once :meth:`close` was called.""" + return self._closed + + async def __aenter__(self) -> "AsyncDatabase": + return self + + async def __aexit__(self, exc_type, exc, tb) -> None: + await self.close() + + async def run(self, fn: Callable[..., Any], /, *args: Any, **kwargs: Any) -> Any: + """Run ``fn(*args, **kwargs)`` on a worker thread and return its result. + + This is the general call: every other method is this one with a database call in ``fn``. Use it for anything the facade does not + wrap, for example ``await adb.run(lambda: db.query("sql", "SELECT count(*) AS n FROM Note").first().get("n"))``. The context + variables of the calling task are visible to ``fn``, as with ``asyncio.to_thread``. An exception of ``fn`` is raised here. + + Raises: + RuntimeError: The facade was closed. + """ + if self._closed: + raise RuntimeError("this AsyncDatabase is closed") + loop = asyncio.get_running_loop() + context = contextvars.copy_context() + call = functools.partial(context.run, fn, *args, **kwargs) + return await loop.run_in_executor(self._executor, call) + + async def query( + self, language: str, command: str, *args: Any + ) -> List[Dict[str, Any]]: + """``db.query(language, command, *args)``, with the rows read on the worker: a list of ``dict`` (``ResultSet.to_list()``). + + The parameters bind as in ``db.query``: positional ``?`` from the extra arguments, or by name from one ``dict``. + """ + return await self.run(self._query, language, command, *args) + + async def command( + self, language: str, command: str, *args: Any + ) -> Optional[List[Dict[str, Any]]]: + """``db.command(language, command, *args)``, with the rows read on the worker: a list of ``dict``, or ``None`` when the + engine returns no result set. + + An ``INSERT``, ``UPDATE`` or ``DELETE`` needs a transaction, and a transaction cannot span awaits: put writes in + :meth:`run_in_transaction`, or switch on ``db.set_auto_transaction(True)``. Do not send ``BEGIN`` or ``COMMIT`` through this + method: the next call may run on another worker. + """ + return await self.run(self._command, language, command, *args) + + async def run_in_transaction( + self, fn: Callable[[], Any], retries: int = 12, backoff_s: float = 0.005 + ) -> Any: + """``db.run_in_transaction(fn, retries, backoff_s)``: ``fn`` runs in one transaction, on one worker, from begin to commit. + + ``fn`` is a plain function with no arguments that makes blocking calls on the ``Database`` (not awaits, and not this facade). + A conflict with another writer rolls the transaction back and runs ``fn`` again, up to ``retries`` times, on the same worker. + An exception of ``fn`` rolls the transaction back and is raised here. Returns what ``fn`` returns. + """ + return await self.run( + self._db.run_in_transaction, fn, retries=retries, backoff_s=backoff_s + ) + + async def find_nearest( + self, + index, + query_vector, + k: int = 10, + ef_search: Optional[int] = None, + allowed_rids=None, + *, + include_vector: bool = True, + ): + """``index.find_nearest(...)`` on a worker: a list of ``(record, distance)``, nearest first. + + ``index`` is the ``VectorIndex`` that ``db.create_vector_index(...)`` or ``db.schema.get_vector_index(...)`` returns. A search + finds every vector whose transaction committed before the search started. + """ + return await self.run( + index.find_nearest, + query_vector, + k=k, + ef_search=ef_search, + allowed_rids=allowed_rids, + include_vector=include_vector, + ) + + async def close(self) -> None: + """Refuse new calls and wait until every call that was submitted has finished, the queued ones included. + + The wait does not block the event loop. Calling it again waits for the same calls. It does not close the ``Database``. + """ + self._closed = True + await asyncio.to_thread(self._executor.shutdown, True) + + def _query(self, language: str, command: str, *args: Any): + return self._db.query(language, command, *args).to_list() + + def _command(self, language: str, command: str, *args: Any): + result = self._db.command(language, command, *args) + return None if result is None else result.to_list() diff --git a/tests/test_aio.py b/tests/test_aio.py new file mode 100644 index 0000000..afce889 --- /dev/null +++ b/tests/test_aio.py @@ -0,0 +1,305 @@ +"""``arcadedb.aio.AsyncDatabase``: the asyncio facade over the blocking calls of a ``Database`` (ArcadeData/arcadedb-python#306). + +The facade runs each call on a worker thread of its own bounded pool, so these tests pin what the docs page ``how-to/asyncio.adoc`` says: +the rows and errors come back to the awaiting task, ``run_in_transaction`` keeps begin, the retries and the commit on one worker and rolls +back on an exception, many tasks can write at once, the pool never runs more than ``max_workers`` calls and does run that many, a cancelled +await abandons the result without stopping a call that started, and ``close()`` waits for the calls that were submitted. The file uses no +Java object, so it runs on the JVM build and on the native build alike. The tests wait on events and barriers, never on the wall clock. +""" + +import asyncio +import contextvars +import threading +import time + +import pytest + +import arcadedb +from arcadedb.aio import AsyncDatabase + +WAIT_S = ( + 30 # a ceiling for an event or a barrier, so a broken test fails instead of hanging +) + + +def _count(db, type_name="Note"): + return db.query("sql", f"SELECT count(*) AS n FROM {type_name}").first().get("n") + + +@pytest.fixture +def notes(temp_db): + temp_db.command("sql", "CREATE DOCUMENT TYPE Note") + return temp_db + + +def test_query_and_command_return_plain_rows_and_take_the_parameters_of_the_sync_calls( + notes, +): + async def main(): + async with AsyncDatabase(notes) as adb: + created = await adb.command("sql", "CREATE DOCUMENT TYPE Other") + assert isinstance(created, list) + await adb.run_in_transaction( + lambda: notes.command("sql", "INSERT INTO Note SET n = ?", 7) + ) + positional = await adb.query("sql", "SELECT n FROM Note WHERE n = ?", 7) + named = await adb.query("sql", "SELECT n FROM Note WHERE n = :n", {"n": 7}) + nothing = await adb.query("sql", "SELECT n FROM Note WHERE n = ?", 8) + return positional, named, nothing + + positional, named, nothing = asyncio.run(main()) + for rows in (positional, named): + assert all(isinstance(row, dict) for row in rows) + assert [row["n"] for row in rows] == [7] + assert nothing == [] + assert notes.is_open(), "closing the facade must not close the database" + + +def test_run_in_transaction_commits_on_one_worker_and_returns_what_the_function_returns( + notes, +): + threads = [] + + def write(): + threads.append(threading.current_thread().name) + notes.command("sql", "INSERT INTO Note SET n = 1") + notes.command("sql", "INSERT INTO Note SET n = 2") + threads.append(threading.current_thread().name) + return "written" + + async def main(): + async with AsyncDatabase(notes, max_workers=4) as adb: + return await adb.run_in_transaction(write) + + assert asyncio.run(main()) == "written" + assert _count(notes) == 2 + assert len(set(threads)) == 1 and threads[0].startswith("arcadedb-aio") + + +def test_an_exception_in_run_in_transaction_rolls_back_and_reaches_the_task(notes): + def write_then_fail(): + notes.command("sql", "INSERT INTO Note SET n = 1") + raise ValueError("boom") + + async def main(): + # one worker: the check after the failure runs on the thread that ran the transaction, where a leftover transaction would be + async with AsyncDatabase(notes, max_workers=1) as adb: + with pytest.raises(ValueError, match="boom"): + await adb.run_in_transaction(write_then_fail) + assert await adb.run(notes.is_transaction_active) is False + assert await adb.run(_count, notes) == 0 + # the worker is usable afterwards + await adb.run_in_transaction( + lambda: notes.command("sql", "INSERT INTO Note SET n = 2") + ) + assert await adb.run(_count, notes) == 1 + + asyncio.run(main()) + + +def test_an_engine_error_is_raised_in_the_awaiting_task(notes): + async def main(): + async with AsyncDatabase(notes) as adb: + with pytest.raises(arcadedb.ArcadeDBError): + await adb.query("sql", "SELEC nothing FROM Note") + # the pool carries on after an error + assert await adb.query("sql", "SELECT count(*) AS n FROM Note") + + asyncio.run(main()) + + +def test_many_tasks_write_and_read_at_once_and_every_write_lands(notes): + tasks = 200 + notes.command("sql", "CREATE DOCUMENT TYPE Counter") + with notes.transaction(): + notes.command("sql", "INSERT INTO Counter SET id = 1, n = 0") + + async def one(adb, i): + def work(): + notes.command("sql", "UPDATE Counter SET n = n + 1 WHERE id = 1") + notes.command("sql", "INSERT INTO Note SET n = ?", i) + + await adb.run_in_transaction(work, retries=60) + rows = await adb.query("sql", "SELECT count(*) AS n FROM Note") + return rows[0]["n"] + + async def main(): + async with AsyncDatabase(notes, max_workers=8) as adb: + return await asyncio.gather(*(one(adb, i) for i in range(tasks))) + + seen = asyncio.run(main()) + assert all(1 <= n <= tasks for n in seen) + assert _count(notes) == tasks + assert _count(notes, "Counter") == 1 + counter = notes.query("sql", "SELECT n FROM Counter WHERE id = 1").first().get("n") + assert counter == tasks, f"{tasks - counter} of {tasks} increments were lost" + values = [row.get("n") for row in notes.query("sql", "SELECT n FROM Note")] + assert sorted(values) == list(range(tasks)) + + +def test_the_pool_runs_max_workers_calls_at_once_and_never_more(): + class Plain: # run() needs no engine call + pass + + workers = 4 + barrier = threading.Barrier(workers) + lock = threading.Lock() + state = {"now": 0, "peak": 0} + + def call(): + with lock: + state["now"] += 1 + state["peak"] = max(state["peak"], state["now"]) + time.sleep(0.01) + with lock: + state["now"] -= 1 + + async def main(): + async with AsyncDatabase(Plain(), max_workers=workers) as adb: + assert adb.max_workers == workers + # all four calls must be inside the barrier together: with fewer running at once the wait times out and raises + await asyncio.gather( + *(adb.run(barrier.wait, WAIT_S) for _ in range(workers)) + ) + await asyncio.gather(*(adb.run(call) for _ in range(5 * workers))) + + asyncio.run(main()) + assert 1 <= state["peak"] <= workers + + +def test_close_waits_for_a_running_call_and_a_closed_facade_refuses_new_calls(): + class Plain: + pass + + started, release = threading.Event(), threading.Event() + + def slow(): + started.set() + assert release.wait(WAIT_S) + return "done" + + async def main(): + adb = AsyncDatabase(Plain(), max_workers=2) + call = asyncio.ensure_future(adb.run(slow)) + assert await asyncio.to_thread(started.wait, WAIT_S) + closing = asyncio.ensure_future(adb.close()) + await asyncio.sleep(0.2) # a close that did not wait would be done long before + assert not closing.done(), "close() returned while a call was still running" + assert adb.closed + with pytest.raises(RuntimeError, match="closed"): + await adb.run(lambda: None) + release.set() + assert await call == "done" + await closing + await adb.close() # a second close has nothing left to wait for + + asyncio.run(main()) + + +def test_close_also_runs_the_calls_that_are_still_queued(): + class Plain: + pass + + started, release = threading.Event(), threading.Event() + ran = [] + + def blocker(): + started.set() + assert release.wait(WAIT_S) + + async def main(): + adb = AsyncDatabase(Plain(), max_workers=1) + first = asyncio.ensure_future(adb.run(blocker)) + assert await asyncio.to_thread(started.wait, WAIT_S) + second = asyncio.ensure_future(adb.run(ran.append, "second")) + await asyncio.sleep( + 0.05 + ) # let the second call reach the queue behind the blocked worker + closing = asyncio.ensure_future(adb.close()) + release.set() + await first + await second + await closing + + asyncio.run(main()) + assert ran == ["second"] + + +def test_cancelling_the_await_abandons_the_result_but_a_started_call_finishes(): + class Plain: + pass + + started, release, finished = ( + threading.Event(), + threading.Event(), + threading.Event(), + ) + + def slow(): + started.set() + assert release.wait(WAIT_S) + finished.set() + + async def main(): + adb = AsyncDatabase(Plain(), max_workers=1) + task = asyncio.ensure_future(adb.run(slow)) + assert await asyncio.to_thread(started.wait, WAIT_S) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + assert not finished.is_set() + release.set() + await adb.close() # waits for the abandoned call + assert finished.is_set() + + asyncio.run(main()) + + +def test_a_call_that_still_waits_for_a_worker_does_not_run_when_its_task_is_cancelled(): + class Plain: + pass + + started, release = threading.Event(), threading.Event() + ran = [] + + def blocker(): + started.set() + assert release.wait(WAIT_S) + + async def main(): + adb = AsyncDatabase(Plain(), max_workers=1) + first = asyncio.ensure_future(adb.run(blocker)) + assert await asyncio.to_thread(started.wait, WAIT_S) + second = asyncio.ensure_future(adb.run(ran.append, "second")) + await asyncio.sleep( + 0.05 + ) # the second call is queued behind the one blocked worker + second.cancel() + with pytest.raises(asyncio.CancelledError): + await second + release.set() + await first + await adb.close() + + asyncio.run(main()) + assert ran == [], "a cancelled call that had not started still ran" + + +def test_the_context_variables_of_the_task_reach_the_call(): + class Plain: + pass + + agent = contextvars.ContextVar("agent") + + async def main(): + async with AsyncDatabase(Plain()) as adb: + agent.set("agent-7") + return await adb.run(agent.get) + + assert asyncio.run(main()) == "agent-7" + + +@pytest.mark.parametrize("bad", [0, -1, True, 2.5, "8", None]) +def test_max_workers_must_be_an_int_of_at_least_one(bad): + with pytest.raises(ValueError, match="max_workers"): + AsyncDatabase(object(), max_workers=bad) diff --git a/tests/test_docs_asyncio_page.py b/tests/test_docs_asyncio_page.py new file mode 100644 index 0000000..cbf12c8 --- /dev/null +++ b/tests/test_docs_asyncio_page.py @@ -0,0 +1,60 @@ +"""Every Python block of the Antora page ``docs/modules/ROOT/pages/how-to/asyncio.adoc`` runs, and prints what the page says. + +Same pattern as ``test_docs_async_page.py``: the blocks are chained on purpose, they run in the order of the page, in one namespace, +against one fresh temporary database that the namespace holds as ``db``. A block that is followed by a ``[source,text]`` block must print +exactly that text. What each statement of the page claims is pinned on its own in ``test_aio.py`` and ``test_vector_read_your_writes.py``. +The test uses no Java object, so it runs on the JVM build and the native build alike. +""" + +import contextlib +import io +import re +from pathlib import Path + +PAGE = ( + Path(__file__).resolve().parents[1] + / "docs" + / "modules" + / "ROOT" + / "pages" + / "how-to" + / "asyncio.adoc" +) + +# a Python listing, then (optionally, directly after it) the text listing with its output +BLOCK_RE = re.compile( + r"\[source,python[^\]\n]*\]\n----\n(?P.*?)\n----\n" + r"(?:\n\[source,text\]\n----\n(?P.*?)\n----\n)?", + re.DOTALL, +) + + +def _blocks(): + text = PAGE.read_text(encoding="utf-8") + found = [(m.group("code"), m.group("out")) for m in BLOCK_RE.finditer(text)] + # no Python listing may escape the extraction (a different spelling of the block header would run nowhere) + assert text.count("[source,python") == len(found) + return found + + +def test_the_page_has_its_blocks(): + blocks = _blocks() + assert len(blocks) >= 5 + assert sum(1 for _, out in blocks if out is not None) >= 5 + + +def test_every_block_runs_in_order_and_prints_what_the_page_says(temp_db): + namespace = {"db": temp_db, "__name__": "asyncio_page"} + for number, (code, expected) in enumerate(_blocks(), start=1): + captured = io.StringIO() + compiled = compile(code, f"asyncio.adoc python block {number}", "exec") + with contextlib.redirect_stdout(captured): + exec(compiled, namespace) # nosec B102 - the page of this repository + if expected is not None: + assert ( + captured.getvalue().strip() == expected.strip() + ), f"block {number} printed other text than the page shows:\n{code}" + else: + assert ( + captured.getvalue() == "" + ), f"block {number} prints, the page shows no output" From fbac4b51c094e31afd2492c0d707a6a2c8a3e92a Mon Sep 17 00:00:00 2001 From: Taewoon Kim Date: Thu, 12 Nov 2026 12:18:06 +0900 Subject: [PATCH 2/3] Read-your-writes tests also through AsyncDatabase; the concurrency and vectors pages point to the facade (refs #306) The two read-your-writes tests of #373 (the writing task, and other tasks that search after the commit) run again through arcadedb.aio.AsyncDatabase: an insert is run_in_transaction and a search is find_nearest, on both wheels. The asyncio section of the concurrency page and the vectors page name the facade next to asyncio.to_thread. Co-Authored-By: Claude Sonnet 5.5 --- .../ROOT/pages/how-to/concurrency.adoc | 2 + docs/modules/ROOT/pages/how-to/vectors.adoc | 2 +- tests/test_vector_read_your_writes.py | 76 ++++++++++++++++++- 3 files changed, 78 insertions(+), 2 deletions(-) diff --git a/docs/modules/ROOT/pages/how-to/concurrency.adoc b/docs/modules/ROOT/pages/how-to/concurrency.adoc index 696b18b..2508f46 100644 --- a/docs/modules/ROOT/pages/how-to/concurrency.adoc +++ b/docs/modules/ROOT/pages/how-to/concurrency.adoc @@ -195,6 +195,8 @@ run_repeatable_read(db, lambda: add_one(3)) Call the bindings from asyncio through `asyncio.to_thread` or `loop.run_in_executor`. A call made directly on the event loop blocks the loop for as long as the call takes. +`arcadedb.aio.AsyncDatabase` does this wrapping for `query`, `command`, `run_in_transaction` and `find_nearest`, on a bounded thread pool of its own: see xref:how-to/asyncio.adoc[asyncio: AsyncDatabase]. The probe below calls the bindings through `asyncio.to_thread` and `run_in_executor` directly. + The probe: 100 tasks, each making 20 operations (10 on Python 3.14t), so 2,000 operations (1,000 on Python 3.14t) per call style. One operation is `db.run_in_transaction(fn, retries=60)` with an atomic counter update and one unique ledger insert, then a count. A heartbeat task sleeps 10 ms in a loop and records how late it wakes up. Each cell below is that lateness in milliseconds, as the 99th percentile and the maximum. In the short phases, where the 99th percentile equals the maximum, there were too few heartbeats for them to differ. The two `async_executor()` rows use the bindings' `AsyncExecutor` with `set_parallel_level(1)` and `set_parallel_level(4)`: each task inserts one row with `command(..., callback=..., error_callback=...)`, and the callbacks hand the result to the loop with `loop.call_soon_threadsafe`. [cols="4,2,2,2"] diff --git a/docs/modules/ROOT/pages/how-to/vectors.adoc b/docs/modules/ROOT/pages/how-to/vectors.adoc index e7c13da..828a51e 100644 --- a/docs/modules/ROOT/pages/how-to/vectors.adoc +++ b/docs/modules/ROOT/pages/how-to/vectors.adoc @@ -508,7 +508,7 @@ print(notes.get_size()) An agent that writes one record per turn and searches right after needs no setting: the rebuild runs in the background and the searches keep running while it does. // Pinned by tests/test_vector_read_your_writes.py, on both packages: 200 inserts, each followed by a search, from the writing task and from other tasks. -A search finds every vector whose transaction committed before the search started. Insert in one call and search in the next, from the same task or from another one, and the new vector is in the result, also while the graph rebuilds in the background. From asyncio, run each call with `asyncio.to_thread` as xref:how-to/concurrency.adoc#asyncio[Concurrency] describes: the insert and the search of one task can run on different threads and the search still finds the vector. This holds for committed writes only. It says nothing about a search made while the writing transaction is still open. +A search finds every vector whose transaction committed before the search started. Insert in one call and search in the next, from the same task or from another one, and the new vector is in the result, also while the graph rebuilds in the background. From asyncio, use `AsyncDatabase` (xref:how-to/asyncio.adoc[asyncio: AsyncDatabase]) or run each call with `asyncio.to_thread` as xref:how-to/concurrency.adoc#asyncio[Concurrency] describes: the insert and the search of one task can run on different threads and the search still finds the vector. This holds for committed writes only. It says nothing about a search made while the writing transaction is still open. If the tail of the search latency matters more than the agent's total time, you can move the rebuild to a moment of your choosing: raise `mutations_before_rebuild` far above the number of writes, set the database setting `arcadedb.vectorIndex.maxDeltaScanRatio` to 0, and call `build_graph_now()` right before a step in which the agent does not search, such as its call to the language model. diff --git a/tests/test_vector_read_your_writes.py b/tests/test_vector_read_your_writes.py index 8a42f90..9d434bc 100644 --- a/tests/test_vector_read_your_writes.py +++ b/tests/test_vector_read_your_writes.py @@ -9,7 +9,8 @@ rebuild threshold is max(100, 20 percent of the graph) = 100 pending changes, and the 200 inserts of a test cross it: the searches meet the delta buffer first and the rebuilt graph later. ``ef_search`` is as large as the corpus on purpose, so a miss cannot be the approximate walk of a large graph: every vector is the nearest neighbour of itself, at distance 0, and a different result means the write was not seen. -The file uses no Java object, so it runs on the JVM build and on the native build alike. +The same two tests run again through ``arcadedb.aio.AsyncDatabase`` (the facade over those calls), where an insert is a ``run_in_transaction`` and +a search is ``find_nearest``. The file uses no Java object, so it runs on the JVM build and on the native build alike. """ import asyncio @@ -20,6 +21,7 @@ import pytest import arcadedb +from arcadedb.aio import AsyncDatabase DIM = 8 BASE = 300 @@ -118,3 +120,75 @@ async def main(): missed == [] ), f"{len(missed)} of {TOTAL} committed vectors were not found by another task: {missed[:5]}" assert _count(db) == BASE + TOTAL + + +def _write(db, i): + db.command("sql", "INSERT INTO Doc SET vid = ?, embedding = ?", i, _vec(i)) + + +async def _facade_insert(adb, db, i): + await adb.run_in_transaction(lambda: _write(db, i), retries=60) + + +async def _facade_nearest_ids(adb, index, i): + hits = await adb.find_nearest(index, _vec(i), k=5, ef_search=EF_SEARCH) + return [record.get("vid") for record, _distance in hits] + + +def test_through_the_facade_a_task_finds_the_vector_it_committed_in_the_call_before( + corpus, +): + db, index = corpus + missed = [] + + async def agent(adb, n): + for step in range(PER_WRITER): + i = BASE + n * PER_WRITER + step + await _facade_insert(adb, db, i) + ids = await _facade_nearest_ids(adb, index, i) + if ids[:1] != [i]: + missed.append((i, ids)) + + async def main(): + async with AsyncDatabase(db, max_workers=8) as adb: + await asyncio.gather(*(agent(adb, n) for n in range(WRITERS))) + + asyncio.run(main()) + assert ( + missed == [] + ), f"{len(missed)} of {TOTAL} committed vectors were not found by the next search: {missed[:5]}" + assert _count(db) == BASE + TOTAL + + +def test_through_the_facade_another_task_finds_the_vector_once_the_writing_task_has_committed_it( + corpus, +): + db, index = corpus + missed = [] + + async def writer(adb, queue): + for step in range(TOTAL): + i = BASE + step + await _facade_insert(adb, db, i) + await queue.put(i) # announced only after the commit returned + for _ in range(READERS): + await queue.put(None) + + async def reader(adb, queue): + while (i := await queue.get()) is not None: + ids = await _facade_nearest_ids(adb, index, i) + if ids[:1] != [i]: + missed.append((i, ids)) + + async def main(): + async with AsyncDatabase(db, max_workers=8) as adb: + queue = asyncio.Queue() + await asyncio.gather( + writer(adb, queue), *(reader(adb, queue) for _ in range(READERS)) + ) + + asyncio.run(main()) + assert ( + missed == [] + ), f"{len(missed)} of {TOTAL} committed vectors were not found by another task: {missed[:5]}" + assert _count(db) == BASE + TOTAL From a9769a4068bdd470416ef6bd3859ab03a285cb2c Mon Sep 17 00:00:00 2001 From: Taewoon Kim Date: Thu, 12 Nov 2026 13:09:36 +0900 Subject: [PATCH 3/3] Test helper: a nosec note for the fixed type name (Bandit B608) The count helper builds its SQL from a fixed type name of the tests, backtick-quoted, with no values. CI's Bandit run flags it at low confidence; the repo's convention is a nosec with the reason. Co-Authored-By: Claude Sonnet 5.5 --- tests/test_aio.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/tests/test_aio.py b/tests/test_aio.py index afce889..b38076f 100644 --- a/tests/test_aio.py +++ b/tests/test_aio.py @@ -23,7 +23,8 @@ def _count(db, type_name="Note"): - return db.query("sql", f"SELECT count(*) AS n FROM {type_name}").first().get("n") + sql = f"SELECT count(*) AS n FROM `{type_name}`" # nosec B608 - a fixed type name of these tests, backtick-quoted, no values + return db.query("sql", sql).first().get("n") @pytest.fixture