From 7bf78db8d0b9e791aad9ee5dfe6d0d4d131026c9 Mon Sep 17 00:00:00 2001 From: Taewoon Kim Date: Sun, 25 Oct 2026 20:47:23 +0900 Subject: [PATCH 1/6] Native: Python functions in SQL and openCypher, record events, Ctrl-C, and the async writer crash Python functions (register_function with replace, deterministic and inferred argument counts, unregister_function) and record events (Database.on) run on the native backend through the existing Java-to-Python upcall. The engine side is a new entry point, arcadedb_n_cb, whose checks, wording, veto and failure rules are those of the JPype bridge. A record argument or event record reaches Python as a Document, Vertex or Edge wrapper. A KeyboardInterrupt in a callback, or one that arrives as the callback is entered, aborts the statement and reaches the caller as a KeyboardInterrupt chained from the ArcadeDBError. A Java thread that runs a Python callback (an async writer) is attached to the isolate already; the thread handle for it is no longer an owned one that detaches the thread when the callback's Python thread state ends. That detach aborted the process in tests/test_async_executor.py (a callback that calls the database from a writer thread). Co-Authored-By: Claude Sonnet 5.5 --- docs/modules/ROOT/pages/installation.adoc | 2 +- .../org/humemai/nativelib/NativeExtras.java | 325 ++++++++++++++++-- src/arcadedb/_native/__init__.py | 2 +- src/arcadedb/_native/database.py | 49 ++- src/arcadedb/_native/extras.py | 215 ++++++++++-- src/arcadedb/_native/interrupts.py | 33 ++ src/arcadedb/_native/lib.py | 11 +- src/arcadedb/callbacks.py | 52 ++- tests/test_native_udf.py | 10 +- tests/test_python_callbacks.py | 13 +- 10 files changed, 624 insertions(+), 88 deletions(-) create mode 100644 src/arcadedb/_native/interrupts.py diff --git a/docs/modules/ROOT/pages/installation.adoc b/docs/modules/ROOT/pages/installation.adoc index b972406..733a505 100644 --- a/docs/modules/ROOT/pages/installation.adoc +++ b/docs/modules/ROOT/pages/installation.adoc @@ -67,7 +67,7 @@ Every wheel is platform-specific and built on a native runner of its platform, s Engine modules the bindings do not use are left out to keep the wheel small. These include the gRPC and MongoDB wire protocols, Raft HA, and Gremlin. -NOTE: *JavaScript is not available.* The GraalJS jars are left out, so `DEFINE FUNCTION ... LANGUAGE js` fails. Define functions in SQL or openCypher. Registering a Python function is planned in https://github.com/ArcadeData/arcadedb-python/pull/29[#29]. +NOTE: *JavaScript is not available.* The GraalJS jars are left out, so `DEFINE FUNCTION ... LANGUAGE js` fails. Define functions in SQL or openCypher, or register a Python function with `Database.register_function` (it works on both packages). // REMOVE WHEN: the old arcadedb-embedded* shims are retired and two releases have shipped with the two-package layout allow-old-name NOTE: *The Studio web UI is in the JVM wheels from 26.11.1.* Studio used to bundle ApexCharts, which is no longer open source (https://github.com/ArcadeData/arcadedb/issues/10579[ArcadeData/arcadedb#10579]). https://github.com/ArcadeData/arcadedb/pull/10627[ArcadeData/arcadedb#10627] replaced it with Apache ECharts. The earlier 26.10.x wheels on PyPI do not contain Studio. The HTTP API of the in-process server works in all of them. diff --git a/native/src/main/java/org/humemai/nativelib/NativeExtras.java b/native/src/main/java/org/humemai/nativelib/NativeExtras.java index 8b86316..ebbddf5 100644 --- a/native/src/main/java/org/humemai/nativelib/NativeExtras.java +++ b/native/src/main/java/org/humemai/nativelib/NativeExtras.java @@ -20,14 +20,27 @@ import com.arcadedb.database.DocumentCallback; import com.arcadedb.database.Identifiable; import com.arcadedb.database.MutableDocument; +import com.arcadedb.database.Record; +import com.arcadedb.database.RecordEvents; import com.arcadedb.database.RID; import com.arcadedb.database.async.DatabaseAsyncExecutor; import com.arcadedb.database.async.ErrorCallback; import com.arcadedb.database.async.NewEdgeCallback; import com.arcadedb.engine.ErrorRecordCallback; import com.arcadedb.engine.WALFile; +import com.arcadedb.event.AfterRecordCreateListener; +import com.arcadedb.event.AfterRecordDeleteListener; +import com.arcadedb.event.AfterRecordUpdateListener; +import com.arcadedb.event.BeforeRecordCreateListener; +import com.arcadedb.event.BeforeRecordDeleteListener; +import com.arcadedb.event.BeforeRecordUpdateListener; +import com.arcadedb.exception.DatabaseOperationException; +import com.arcadedb.function.CypherBuiltinFunctions; +import com.arcadedb.function.CypherFunctionRegistry; +import com.arcadedb.function.FunctionRegistry; import com.arcadedb.function.sql.DefaultSQLFunctionFactory; import com.arcadedb.function.sql.SQLFunctionAbstract; +import com.arcadedb.graph.Edge; import com.arcadedb.graph.GraphBatch; import com.arcadedb.graph.MutableVertex; import com.arcadedb.graph.Vertex; @@ -35,6 +48,9 @@ import com.arcadedb.integration.importer.Importer; import com.arcadedb.query.sql.executor.CommandContext; import com.arcadedb.query.sql.executor.ResultSet; +import com.arcadedb.schema.DocumentType; +import com.arcadedb.schema.EdgeType; +import com.arcadedb.schema.VertexType; import org.graalvm.nativeimage.IsolateThread; import org.graalvm.nativeimage.c.function.CEntryPoint; import org.graalvm.nativeimage.c.function.CFunctionPointer; @@ -240,7 +256,9 @@ public static CCharPointer gbCreate(final IsolateThread thread, final long dbh, try { final Database db = ArcadeNative.db(dbh); final Map o = args(options); - GraphBatch.Builder b = GraphBatch.builder(db); + // db.batch(), as the JPype path: it hands the builder the engine's single-batch guard (LocalDatabase.batchStarted, taken by build() and + // released by close()), which GraphBatch.builder(db) bypasses; a second batch while one is open is refused instead of losing edges (P-04) + GraphBatch.Builder b = db.batch(); for (final Map.Entry e : o.entrySet()) { final Object v = e.getValue(); final int iv = intOf(v, 0); @@ -385,6 +403,19 @@ private static Object[] objects(final Object list) { return ((List) list).toArray(); } + /** + * Whether a bulk load into the type creates vertices (true) or documents (false). + * + * @throws IllegalArgumentException for an edge type: an edge cannot be created without its two endpoints + */ + private static boolean isVertexType(final Database db, final String typeName) { + final DocumentType type = db.getSchema().getType(typeName); + if (type instanceof EdgeType) + throw new IllegalArgumentException("Cannot bulk-insert into '" + typeName + + "' because it is an edge type; load the vertices, then create the edges (graph_batch().new_edges() or CREATE EDGE)"); + return type instanceof VertexType; + } + /** * ops: get | set {key,value} | wait {timeout_ms} | kill | create_record {type, kind, props, cb, err} | command / query {lang, text, params, * cb, err} | on_ok {cb} | on_error {cb} | append_samples {type, ts, cols} | insert_parallel {type, rows} | failures {fid}. @@ -539,8 +570,12 @@ public void onError(final Exception e) { }; final String type = (String) a.get("type"); try { - for (final Object row : (List) a.get("rows")) { - final MutableDocument doc = db.newDocument(type); + final List rows = (List) a.get("rows"); + // the type is looked up once per call, as DocumentBatcher does on the JPype path (humemai #519): a vertex type gets vertices, + // a document type documents, an edge type is refused (an edge needs both endpoints); a missing type raises the schema's own error + final boolean vertex = !rows.isEmpty() && isVertexType(db, type); + for (final Object row : rows) { + final MutableDocument doc = vertex ? db.newVertex(type) : db.newDocument(type); for (final Map.Entry e : ((Map) row).entrySet()) doc.set(e.getKey(), e.getValue()); ex.createRecord(doc, null, onError); @@ -647,39 +682,109 @@ public static CCharPointer exportDatabase(final IsolateThread thread, final long } } - // ---- Python functions in SQL: arcadedb_n_udf_register(name, callbackId, minArgs, maxArgs) + // ---- Python functions in SQL and record events: arcadedb_n_cb(op, json) + + /* + * One entry point for the callback registry of the Python side (functions and record events), JSON in and out like the + * rest: {"ok":true,..} or {"ok":true,"refused":"text"} when the request is understood but not allowed (Python raises the + * text as an ArcadeDBError). The engine logic is the one of the JPype bridge (src/java/com/arcadedb/python/PythonFunctions + * and PythonRecordEvents): same checks, same wording, same veto and failure rules; only the call into Python differs (the + * upcall, with JSON). Keep the two in step. + * + * ops: fn_register {name,cb,min,max,det,replace} fn_unregister {h} ev_register {db,type,event,cb} ev_remove {h} + * A handle (h) names a registration for the unregister call; the reply to a replace or an unregister carries the callback id + * of the registration that went away ("dropped"), so Python can forget it. + */ + + private static final Object REGISTRY = new Object(); + private static final Map HANDLES = new ConcurrentHashMap<>(); + private static final AtomicLong NEXT_HANDLE = new AtomicLong(1); + + static String describe(final Throwable t) { + final String message = t.getMessage(); + return message == null || message.isEmpty() ? t.getClass().getName() : message; + } + + /** The reply of a Python call that must return a value: a missing reply or an {"error"} reply fails with Python's text. */ + @SuppressWarnings("unchecked") + private static Map answer(final String reply) { + if (reply == null) + throw new IllegalStateException("the call was interrupted before Python returned a value"); + final Map r = (Map) Json.parse(reply); + if (r.get("error") != null) + throw new IllegalStateException(String.valueOf(r.get("error"))); + return r; + } + + /** A record as a tagged dict with its category, for Python to wrap as a Document, Vertex or Edge. */ + static void appendRecord(final StringBuilder sb, final Document doc) { + sb.append('{'); + if (doc.getIdentity() != null) { + sb.append("\"@rid\":"); + NativeCodec.quote(sb, doc.getIdentity().toString()); + sb.append(','); + } + sb.append("\"@type\":"); + NativeCodec.quote(sb, doc.getTypeName()); + sb.append(",\"@cat\":\"").append(doc instanceof Vertex ? 'v' : doc instanceof Edge ? 'e' : 'd').append('"'); + for (final String name : doc.getPropertyNames()) { + sb.append(','); + NativeCodec.quote(sb, name); + sb.append(':'); + NativeCodec.append(sb, doc.get(name)); + } + sb.append('}'); + } /** * A SQL function whose body is a Python callable. The engine thread that runs the statement calls the upcall with - * {"k":"fn","args":[..]} (tagged values, as parameters) and expects {"v":} or {"error":"text"}; an error is thrown - * as a CommandExecutionException from the function, so the statement fails with the Python message. + * {"k":"fn","args":[..]} (tagged values as parameters, a record argument as a dict with @rid, @type and @cat) and expects + * {"v":} or {"error":"text"}; any failure is thrown as a CommandExecutionException from the function, so the + * statement fails with the Python message. */ static final class PythonFunction extends SQLFunctionAbstract { - private final long cb; + final long handle = NEXT_HANDLE.getAndIncrement(); + final long cb; private final int min, max; + private final boolean deterministic; + volatile boolean revoked; - PythonFunction(final String name, final long cb, final int min, final int max) { + PythonFunction(final String name, final long cb, final int min, final int max, final boolean deterministic) { super(name); this.cb = cb; this.min = min; this.max = max; + this.deterministic = deterministic; } @Override public Object execute(final Object self, final Identifiable record, final Object result, final Object[] params, final CommandContext ctx) { - final StringBuilder sb = new StringBuilder("{\"k\":\"fn\",\"args\":["); - for (int i = 0; i < params.length; i++) { - if (i > 0) - sb.append(','); - NativeCodec.append(sb, params[i]); + final String name = getName(); + if (revoked) + throw new com.arcadedb.exception.CommandExecutionException("Python function '" + name + "' was unregistered"); + if (params.length < min || params.length > max) + throw new com.arcadedb.exception.CommandExecutionException( + "Python function '" + name + "' takes " + arity() + " argument(s), got " + params.length); + try { + final StringBuilder sb = new StringBuilder("{\"k\":\"fn\",\"args\":["); + for (int i = 0; i < params.length; i++) { + if (i > 0) + sb.append(','); + if (params[i] instanceof Document doc) + appendRecord(sb, doc); + else + NativeCodec.append(sb, params[i]); + } + return answer(upcall(cb, sb.append("]}").toString())).get("v"); + } catch (final Throwable t) { + throw new com.arcadedb.exception.CommandExecutionException("Python function '" + name + "' failed: " + describe(t), t); } - final String reply = upcall(cb, sb.append("]}").toString()); - if (reply == null) - return null; - @SuppressWarnings("unchecked") final Map r = (Map) Json.parse(reply); - if (r.get("error") != null) - throw new com.arcadedb.exception.CommandExecutionException(String.valueOf(r.get("error"))); - return r.get("v"); + } + + private String arity() { + if (min == max) + return String.valueOf(min); + return max == Integer.MAX_VALUE ? "at least " + min : min + " to " + max; } @Override @@ -692,17 +797,187 @@ public int getMaxArgs() { return max; } + @Override + public boolean isDeterministic() { + return deterministic; + } + @Override public String getSyntax() { - return getName() + "(*)"; + return getName() + "(" + (max == 0 ? "" : "") + ") [Python]"; + } + } + + private static boolean takenByEngine(final String key) { + final DefaultSQLFunctionFactory factory = DefaultSQLFunctionFactory.getInstance(); + if (factory.hasFunction(key)) + return !(factory.getFunctionInstance(key) instanceof PythonFunction); + return CypherBuiltinFunctions.isBuiltin(key) || CypherFunctionRegistry.hasFunction(key) || FunctionRegistry.hasFunction(key); + } + + private static PythonFunction pythonFunction(final String key) { + final DefaultSQLFunctionFactory factory = DefaultSQLFunctionFactory.getInstance(); + return factory.hasFunction(key) && factory.getFunctionInstance(key) instanceof PythonFunction f ? f : null; + } + + /** Revokes and unregisters a function; the callback id it had, or 0 when it was not the registered one any more. */ + private static long removeFunction(final PythonFunction function) { + function.revoked = true; + HANDLES.remove(function.handle); + final String key = function.getName(); + final DefaultSQLFunctionFactory factory = DefaultSQLFunctionFactory.getInstance(); + if (factory.hasFunction(key) && factory.getFunctionInstance(key) == function) { + factory.unregister(key); + FunctionRegistry.unregister(key); + } + return function.cb; + } + + /** The listener of one event of a Python callback: a before_* callback may veto, any failure fails the operation. */ + static final class PythonListener implements BeforeRecordCreateListener, BeforeRecordUpdateListener, BeforeRecordDeleteListener, + AfterRecordCreateListener, AfterRecordUpdateListener, AfterRecordDeleteListener { + final long handle = NEXT_HANDLE.getAndIncrement(); + final long cb; + final String event; + final RecordEvents events; + + PythonListener(final RecordEvents events, final String event, final long cb) { + this.events = events; + this.event = event; + this.cb = cb; + } + + private boolean dispatch(final Record record) { + if (!(record instanceof Document doc)) + return true; + try { + final StringBuilder sb = new StringBuilder("{\"k\":\"ev\",\"e\":\"").append(event).append("\",\"rec\":"); + appendRecord(sb, doc); + return !Boolean.FALSE.equals(answer(upcall(cb, sb.append('}').toString())).get("v")); + } catch (final Throwable t) { + throw new DatabaseOperationException("Python '" + event + "' callback failed: " + describe(t), t); + } } + + @Override + public boolean onBeforeCreate(final Record record) { + return dispatch(record); + } + + @Override + public boolean onBeforeUpdate(final Record record) { + return dispatch(record); + } + + @Override + public boolean onBeforeDelete(final Record record) { + return dispatch(record); + } + + @Override + public void onAfterCreate(final Record record) { + dispatch(record); + } + + @Override + public void onAfterUpdate(final Record record) { + dispatch(record); + } + + @Override + public void onAfterDelete(final Record record) { + dispatch(record); + } + + void register() { + switch (event) { + case "before_create" -> events.registerListener((BeforeRecordCreateListener) this); + case "before_update" -> events.registerListener((BeforeRecordUpdateListener) this); + case "before_delete" -> events.registerListener((BeforeRecordDeleteListener) this); + case "after_create" -> events.registerListener((AfterRecordCreateListener) this); + case "after_update" -> events.registerListener((AfterRecordUpdateListener) this); + case "after_delete" -> events.registerListener((AfterRecordDeleteListener) this); + default -> throw new IllegalArgumentException("Unknown record event: " + event); + } + } + + void unregister() { + switch (event) { + case "before_create" -> events.unregisterListener((BeforeRecordCreateListener) this); + case "before_update" -> events.unregisterListener((BeforeRecordUpdateListener) this); + case "before_delete" -> events.unregisterListener((BeforeRecordDeleteListener) this); + case "after_create" -> events.unregisterListener((AfterRecordCreateListener) this); + case "after_update" -> events.unregisterListener((AfterRecordUpdateListener) this); + case "after_delete" -> events.unregisterListener((AfterRecordDeleteListener) this); + default -> throw new IllegalArgumentException("Unknown record event: " + event); + } + } + } + + private static CCharPointer refused(final String text) { + final StringBuilder sb = new StringBuilder("{\"ok\":true,\"refused\":"); + NativeCodec.quote(sb, text); + return ArcadeNative.out(sb.append('}').toString()); + } + + private static long longOf(final Object v) { + return ((Number) v).longValue(); } - @CEntryPoint(name = "arcadedb_n_udf_register") - public static CCharPointer udfRegister(final IsolateThread thread, final CCharPointer name, final long cb, final int min, final int max) { + @CEntryPoint(name = "arcadedb_n_cb") + public static CCharPointer callbackRegistry(final IsolateThread thread, final CCharPointer op, final CCharPointer json) { try { - DefaultSQLFunctionFactory.getInstance().register(new PythonFunction(ArcadeNative.str(name), cb, min, max)); - return ArcadeNative.OK(); + final Map a = args(json); + switch (ArcadeNative.str(op)) { + case "fn_register": { + final String name = String.valueOf(a.get("name")); + final String key = name.toLowerCase(java.util.Locale.ENGLISH); + synchronized (REGISTRY) { + if (takenByEngine(key)) + return refused("Cannot register '" + name + "': it is already a function of the engine"); + final PythonFunction old = pythonFunction(key); + if (old != null && !Boolean.TRUE.equals(a.get("replace"))) + return refused("A Python function '" + name + "' is already registered (pass replace=True to replace it)"); + final long dropped = old == null ? 0 : removeFunction(old); + final PythonFunction function = new PythonFunction(key, longOf(a.get("cb")), (int) longOf(a.get("min")), (int) longOf(a.get("max")), + Boolean.TRUE.equals(a.get("det"))); + DefaultSQLFunctionFactory.getInstance().register(function); + HANDLES.put(function.handle, function); + return ArcadeNative.out("{\"ok\":true,\"h\":" + function.handle + ",\"dropped\":" + dropped + "}"); + } + } + case "fn_unregister": { + synchronized (REGISTRY) { + final Object handle = HANDLES.get(longOf(a.get("h"))); + final long dropped = handle instanceof PythonFunction f ? removeFunction(f) : 0; + return ArcadeNative.out("{\"ok\":true,\"dropped\":" + dropped + "}"); + } + } + case "ev_register": { + final Database db = ArcadeNative.db(longOf(a.get("db"))); + final String type = (String) a.get("type"); + final RecordEvents events; + if (type == null) + events = db.getEvents(); + else { + if (!db.getSchema().existsType(type)) + return refused("Type '" + type + "' does not exist"); + events = db.getSchema().getType(type).getEvents(); + } + final PythonListener listener = new PythonListener(events, String.valueOf(a.get("event")), longOf(a.get("cb"))); + listener.register(); + HANDLES.put(listener.handle, listener); + return ArcadeNative.out("{\"ok\":true,\"h\":" + listener.handle + "}"); + } + case "ev_remove": { + final Object handle = HANDLES.remove(longOf(a.get("h"))); + if (handle instanceof PythonListener listener) + listener.unregister(); + return ArcadeNative.out("{\"ok\":true,\"dropped\":" + (handle instanceof PythonListener l ? l.cb : 0) + "}"); + } + default: + throw new IllegalArgumentException("unknown callback op " + ArcadeNative.str(op)); + } } catch (final Throwable t) { return ArcadeNative.err(t); } diff --git a/src/arcadedb/_native/__init__.py b/src/arcadedb/_native/__init__.py index 27820cb..b413293 100644 --- a/src/arcadedb/_native/__init__.py +++ b/src/arcadedb/_native/__init__.py @@ -5,7 +5,7 @@ library (``libarcadedb``) that Python loads with ``ctypes``: no JVM, no JPype. The surface is the DSL-first core: databases, ``query``/``command`` for SQL and openCypher, transactions, results and the in-process server. -What is not on native yet (each raises :class:`NotOnNativeYet`): Python UDFs and record callbacks, ``GraphBatch``, the +What is not on native yet (each raises :class:`NotOnNativeYet`): ``GraphBatch``, the async executor, the Java-object APIs (``Document``/``Vertex``/``Edge`` wrappers, ``new_vertex``, ``new_document``, ``schema``, ``get_java_database``), the columnar/Arrow readers other than a simple row conversion, importers and exporters that go through Java objects, and every JVM setting (``jvm_kwargs``, ``start_jvm``). diff --git a/src/arcadedb/_native/database.py b/src/arcadedb/_native/database.py index d640e8a..9646cbb 100644 --- a/src/arcadedb/_native/database.py +++ b/src/arcadedb/_native/database.py @@ -13,7 +13,7 @@ from ..exceptions import ArcadeDBError from ..transactions import TransactionContext from ..type_conversion import json_bulk_dumps -from . import lib, not_yet +from . import interrupts, lib, not_yet from .codec import _prep as _prep_row from .codec import encode_params, json_native, loads, row_default @@ -39,7 +39,9 @@ def _error(prefix: str, reply: dict) -> ArcadeDBError: exc = ArcadeDBError(f"{prefix}: {head}" if prefix else head) exc.native = reply exc.java_chain = chain - return exc + return interrupts.convert( + exc + ) # a Ctrl-C in a Python function is a KeyboardInterrupt chained from this error def _read_error(reply: dict) -> ArcadeDBError: @@ -48,7 +50,7 @@ def _read_error(reply: dict) -> ArcadeDBError: ) exc.native = reply exc.java_chain = reply.get("chain") or [] - return exc + return interrupts.convert(exc) def _call(prefix: str, name: str, *args) -> dict: @@ -392,6 +394,7 @@ def __init__(self, handle: int, owned_by_server: bool = False): self._closed = False self._server_owned = owned_by_server self._async_executors: list = [] + self._registry = None # Python functions and event callbacks, once one is registered (extras.Registrations) self._vector_meta: dict = {} self._vector_dimensions: dict = {} self._vector_id_properties: dict = {} @@ -412,6 +415,8 @@ def close(self): if self._closed: return self._close_async_executors() + if self._registry is not None: + self._registry.clear() # before the engine closes: the functions are process-wide, the listeners die with the database if self._server_owned: # owned by the server: the engine refuses the close, and this handle is done either way self._closed = True @@ -481,6 +486,7 @@ def _exec( refuse_surplus_positionals( "command" if flags & 1 else "query", language, command, args ) + interrupts.clear() if flags & 1 and _LAZY_COMMAND_ROWS and lib.CMD_PATH: # phase 8: the cursor id comes back as a long (no reply text to free and decode); the rows are read late or never text, buf, count = encode_params(args, binary=True) @@ -876,13 +882,40 @@ def async_executor(self): self._async_executors.append(executor) return executor - def register_function(self, name, fn, min_args=0, max_args=-1): - """Make a Python callable available to SQL (process-wide, see ``extras.register_function``).""" - from .extras import register_function + def _registrations(self): + from .extras import Registrations - register_function(name, fn, min_args, max_args) + if self._registry is None: + self._registry = Registrations(self) + return self._registry + + def register_function( + self, + name, + fn, + *, + min_args=None, + max_args=None, + deterministic=False, + replace=False, + ): + """Make a Python callable available to SQL and openCypher (``Database.register_function``: process-wide, the Python + function is called from the engine thread that runs the statement).""" + self._check_not_closed() + self._registrations().register_function( + name, fn, min_args, max_args, deterministic, replace + ) + + def unregister_function(self, name): + """Remove a Python function registered through this database.""" + self._check_not_closed() + self._registrations().unregister_function(name) + + def on(self, event, fn, type_name=None): + """Call a Python function when a record is created, updated or deleted (``Database.on``).""" + self._check_not_closed() + return self._registrations().on(event, fn, type_name) - @property def schema(self): """DDL helpers over SQL statements (the DSL-first subset); everything else is not on native yet.""" self._check_not_closed() diff --git a/src/arcadedb/_native/extras.py b/src/arcadedb/_native/extras.py index 69c6789..96873ef 100644 --- a/src/arcadedb/_native/extras.py +++ b/src/arcadedb/_native/extras.py @@ -1,6 +1,6 @@ """Native backend coverage beyond the DSL core: GraphBatch, the async executor, database settings, import/export and Python functions in SQL, over the ``arcadedb_n_gb*``, ``arcadedb_n_async``, ``arcadedb_n_dbset``, ``arcadedb_n_import``, -``arcadedb_n_export`` and ``arcadedb_n_udf_register`` entry points (``NativeExtras.java``). +``arcadedb_n_export`` and ``arcadedb_n_cb`` entry points (``NativeExtras.java``). Callbacks (async completions, error handlers, Python functions called from SQL) go through ONE ctypes function pointer that ``_ensure_upcall`` registers with the library: the engine thread (or an async writer thread) calls ``_dispatch(id, json)``, @@ -10,13 +10,17 @@ import ctypes import itertools import json +import sys import threading +import weakref from collections.abc import Mapping from typing import Any, Callable, Iterable, Optional +from ..callbacks import EventHandle as _BaseEventHandle +from ..callbacks import check_event, check_function from ..exceptions import ArcadeDBError from ..graph_batch import GraphBatch as _BaseGraphBatch -from . import lib, not_yet +from . import interrupts, lib, not_yet from .codec import _PLAIN, _default, _prep, loads # -------------------------------------------------------------------------------------------------------- upcall @@ -43,6 +47,7 @@ def _dispatch(cid, raw): reply = fn(message) if fn is not None else None data = None if reply is None else _enc(reply).encode("utf-8") except BaseException as exc: # nosec B110 - an exception must not cross the C frame + interrupts.note(exc) data = json.dumps({"error": f"{type(exc).__name__}: {exc}"}).encode("utf-8") if data is None: return None @@ -52,10 +57,32 @@ def _dispatch(cid, raw): return ctypes.addressof(buf) +def _install_unraisable_hook() -> None: + """Keep a Ctrl-C that arrives as the callback is entered. + + CPython runs a pending signal handler at the first bytecode of a function, so a KeyboardInterrupt can be raised before + ``_dispatch`` reaches its ``try``: ctypes then reports it as unraisable and returns NULL to the engine. The hook records it + as the interrupt of this thread (the engine fails the statement on a NULL reply, and the error becomes the + KeyboardInterrupt); every other unraisable exception goes to the hook that was there. + """ + previous = sys.unraisablehook + + def hook(unraisable): + if unraisable.object is _dispatch and isinstance( + unraisable.exc_value, KeyboardInterrupt + ): + interrupts.note(unraisable.exc_value) + else: + previous(unraisable) + + sys.unraisablehook = hook + + def _ensure_upcall() -> None: global _upcall_ref if _upcall_ref is not None: return + _install_unraisable_hook() lib.call_void( "arcadedb_n_set_upcall", ctypes.cast(_make_pointer(), ctypes.c_void_p) ) @@ -67,6 +94,13 @@ def _make_pointer(): return _upcall_ref +def _forget(cid) -> None: + """Drop a callback the engine no longer holds (an unregistered function, a removed listener, a replaced function).""" + if cid: + with _callbacks_lock: + _callbacks.pop(cid, None) + + def _register(fn: Callable) -> int: _ensure_upcall() with _callbacks_lock: @@ -789,31 +823,164 @@ def export_database( )["stats"] -# -------------------------------------------------------------------------------------------------------- Python functions in SQL +# -------------------------------------------------------------------------------------------------------- Python functions in SQL and record events +_MAX_LONG = 2**63 -def register_function( - name: str, fn: Callable, min_args: int = 0, max_args: int = -1 -) -> None: - """Make ``fn`` callable from SQL as ``name(..)``: ``SELECT name(a, b) FROM T``. - ``fn`` receives the argument values as Python values and returns a value of a type a parameter can carry (str, int, float, - bool, None, list, dict, Decimal, datetime). An exception raised by ``fn`` fails the statement with its text. Registration is - process-wide (the engine's function factory), like any engine function; ``max_args`` < 0 means any number. - """ +def _check_result(value) -> None: + """A function result the engine can carry: an int past a Java long is an OverflowError, as on the JPype path.""" + t = type(value) + if t is int: + if not -_MAX_LONG <= value < _MAX_LONG: + raise OverflowError("Python int too large to convert to Java long") + elif t is list or t is tuple: + for item in value: + _check_result(item) + elif t is dict: + for item in value.values(): + _check_result(item) + + +def _record_or_value(database, value): + """A record the engine sent as a dict with ``@rid``, ``@type`` and ``@cat`` becomes a Document, Vertex or Edge.""" + if type(value) is dict and "@type" in value and "@cat" in value: + from .records import _wrap_dict + + return _wrap_dict(database, value) + return value + + +def _function_handler(fn: Callable, database_ref) -> Callable: + def handler(message): + database = database_ref() + result = fn(*[_record_or_value(database, a) for a in message.get("args", [])]) + _check_result(result) + return {"v": result} + + return handler + + +def _event_handler(event: str, fn: Callable, database_ref) -> Callable: + is_before = event.startswith("before_") def handler(message): + from .records import _wrap_dict + + result = fn(_wrap_dict(database_ref(), message["rec"])) + return {"v": not (is_before and result is False)} + + return handler + + +class NativeEventHandle(_BaseEventHandle): + """A registered record event callback; ``remove()`` stops it (also a context manager, as on the JPype backend).""" + + def __init__(self, registrations, event, type_name, handle, cid): + super().__init__(None, event, type_name, handle, None) + self._registrations = registrations + self._cid = cid + + def remove(self) -> None: + if not self._active: + return + self._active = False + self._registrations._events.discard(self) try: - return {"v": fn(*message.get("args", []))} - except Exception as exc: - return {"error": f"{type(exc).__name__}: {exc}"} - - cid = _register(handler) - _call( - "Failed to register the function", - "arcadedb_n_udf_register", - lib.b(name), - cid, - int(min_args), - int(max_args if max_args >= 0 else 2**31 - 1), - ) + _call( + f"Failed to remove '{self.event}' callback", + "arcadedb_n_cb", + lib.b("ev_remove"), + lib.b(json.dumps({"h": self._java_handle})), + ) + finally: + _forget(self._cid) + + +class Registrations: + """The Python functions and event callbacks one ``NativeDatabase`` registered, to remove them when it closes. + + The engine keeps functions for the whole process, so each database removes its own; the checks and the replies are those + of the JPype backend (``arcadedb.callbacks``), the engine side is ``arcadedb_n_cb`` in ``NativeExtras.java``. + """ + + def __init__(self, database): + # weak: the callbacks hold the database only for wrapping records, and the registry must not keep it alive + self._database_ref = weakref.ref(database) + self._h = database._h + self._functions: dict = {} # lower-case name -> (engine handle, callback id) + self._events: set = set() + + def _cb(self, prefix: str, op: str, **args) -> dict: + reply = _call(prefix, "arcadedb_n_cb", lib.b(op), lib.b(json.dumps(args))) + if "refused" in reply: + raise ArcadeDBError(reply["refused"]) + return reply + + def register_function( + self, name, fn, min_args, max_args, deterministic, replace + ) -> None: + min_args, max_args = check_function(name, fn, min_args, max_args) + cid = _register(_function_handler(fn, self._database_ref)) + try: + reply = self._cb( + f"Failed to register function '{name}'", + "fn_register", + name=name, + cb=cid, + min=min_args, + max=min(max_args, 2**31 - 1), + det=bool(deterministic), + replace=bool(replace), + ) + except BaseException: + _forget(cid) + raise + _forget(reply.get("dropped")) # the function this one replaced + self._functions[name.lower()] = (reply["h"], cid) + + def unregister_function(self, name: str) -> None: + entry = self._functions.pop(str(name).lower(), None) + if entry is None: + raise ArcadeDBError( + f"No Python function '{name}' is registered through this database" + ) + try: + self._cb(f"Failed to unregister '{name}'", "fn_unregister", h=entry[0]) + finally: + _forget(entry[1]) + + def on(self, event: str, fn: Callable, type_name: Optional[str]): + check_event(event, fn) + cid = _register(_event_handler(event, fn, self._database_ref)) + try: + reply = self._cb( + f"Failed to register '{event}' callback", + "ev_register", + db=self._h, + type=type_name, + event=event, + cb=cid, + ) + except BaseException: + _forget(cid) + raise + handle = NativeEventHandle(self, event, type_name, reply["h"], cid) + self._events.add(handle) + return handle + + def clear(self) -> None: + """Remove everything this database registered (called when it closes); a failure here must not stop the close.""" + functions, self._functions = self._functions, {} + events, self._events = self._events, set() + for handle, cid in functions.values(): + try: + self._cb("", "fn_unregister", h=handle) + except Exception: # nosec B110 - the library may be shutting down + pass + _forget(cid) + for event in events: + try: + event.remove() + except Exception: # nosec B110 + pass diff --git a/src/arcadedb/_native/interrupts.py b/src/arcadedb/_native/interrupts.py new file mode 100644 index 0000000..c3bfcaa --- /dev/null +++ b/src/arcadedb/_native/interrupts.py @@ -0,0 +1,33 @@ +"""Ctrl-C inside a Python function or callback that the engine called, on the native backend. + +A ``KeyboardInterrupt`` cannot cross the engine as it is: it ends the statement as an engine error. It is remembered per +thread when the upcall catches it (``note``), and the place that turns the statement's failure into an ``ArcadeDBError`` +raises the ``KeyboardInterrupt`` instead (chained from that error, ``convert``), so Ctrl-C reaches the caller as a +``KeyboardInterrupt`` wherever the signal landed: in the function, in the engine between two calls, or on the way into the +callback (the same rule as the JPype backend, see ``arcadedb.callbacks``). +""" + +import threading + +_PENDING = threading.local() + + +def note(error: BaseException) -> None: + """Remember a ``KeyboardInterrupt`` raised in a callback on this thread.""" + if isinstance(error, KeyboardInterrupt): + _PENDING.interrupt = error + + +def clear() -> None: + """Forget an interrupt an earlier statement left (called when a statement starts).""" + _PENDING.interrupt = None + + +def convert(error: BaseException) -> BaseException: + """The ``KeyboardInterrupt`` to raise in place of ``error`` if a callback left one on this thread, else ``error``.""" + interrupt = getattr(_PENDING, "interrupt", None) + if interrupt is None: + return error + _PENDING.interrupt = None + interrupt.__cause__ = error + return interrupt diff --git a/src/arcadedb/_native/lib.py b/src/arcadedb/_native/lib.py index b2f16ac..d0c05d5 100644 --- a/src/arcadedb/_native/lib.py +++ b/src/arcadedb/_native/lib.py @@ -146,6 +146,8 @@ class _Params(ctypes.Structure): ctypes.POINTER(ctypes.c_void_p), ] lib.graal_attach_thread.restype = ctypes.c_int + lib.graal_get_current_thread.argtypes = [ctypes.c_void_p] + lib.graal_get_current_thread.restype = ctypes.c_void_p lib.graal_detach_thread.argtypes = [ctypes.c_void_p] lib.graal_detach_thread.restype = ctypes.c_int # a Java long is 64 bits on every platform; C long is 32 bits on Windows, so L is long long (the same size as c_long elsewhere) @@ -176,7 +178,7 @@ class _Params(ctypes.Structure): "arcadedb_n_async": [P, L, S, S], "arcadedb_n_import": [P, L, S], "arcadedb_n_export": [P, L, S, S, I, S], - "arcadedb_n_udf_register": [P, S, L, I, I], + "arcadedb_n_cb": [P, S, S], "arcadedb_n_set_upcall": [P, P], "arcadedb_free": [P, P], } @@ -222,6 +224,13 @@ def loaded() -> bool: def _thread_handle(): t = getattr(_local, "thread", None) if t is None: + # A Java thread that runs a Python callback (an async writer, an engine thread) is attached to the isolate already. + # Attaching it again would hand Python an owned handle that detaches the thread, while it still runs Java code, as + # soon as the callback's Python thread state ends: the process aborts. + current = _lib.graal_get_current_thread(_isolate) + if current: + t = _local.thread = _Thread(ctypes.c_void_p(current), owned=False) + return t.handle thread = ctypes.c_void_p() rc = _lib.graal_attach_thread(_isolate, ctypes.byref(thread)) if rc != 0: diff --git a/src/arcadedb/callbacks.py b/src/arcadedb/callbacks.py index 62274ac..17904a8 100644 --- a/src/arcadedb/callbacks.py +++ b/src/arcadedb/callbacks.py @@ -153,6 +153,38 @@ def _arity(fn: Callable) -> Tuple[int, int]: return minimum, maximum +def check_function( + name: str, fn: Callable, min_args: Optional[int], max_args: Optional[int] +) -> Tuple[int, int]: + """Validate the arguments of ``register_function`` and return the (min, max) argument counts to register. + + Shared by both backends (the JPype ``FunctionRegistration`` and the native one), so a bad call fails the same way. + """ + if not isinstance(name, str) or not _NAME_RE.match(name): + raise ValueError( + f"Invalid function name {name!r}: use letters, digits and underscores " + "(not starting with a digit), optionally joined by dots like 'py.slugify'" + ) + if not callable(fn): + raise TypeError("fn must be callable") + found_min, found_max = _arity(fn) + min_args = found_min if min_args is None else min_args + max_args = found_max if max_args is None else max_args + if min_args < 0 or max_args < min_args: + raise ValueError( + f"Invalid argument counts: min_args={min_args}, max_args={max_args}" + ) + return min_args, max_args + + +def check_event(event: str, fn: Callable) -> None: + """Validate the arguments of ``Database.on`` (shared by both backends).""" + if event not in EVENTS: + raise ValueError(f"Invalid event {event!r}. Must be one of: {list(EVENTS)}") + if not callable(fn): + raise TypeError("fn must be callable") + + # The proxy classes are defined once, on first use (the JVM must be running, so not at import): # a @JImplements class defined on every registration leaked about 6 KB of Python heap per # registration, which neither the engine's unregister nor gc gave back. State is per instance. @@ -273,20 +305,7 @@ def register( deterministic: bool, replace: bool, ) -> None: - if not isinstance(name, str) or not _NAME_RE.match(name): - raise ValueError( - f"Invalid function name {name!r}: use letters, digits and underscores " - "(not starting with a digit), optionally joined by dots like 'py.slugify'" - ) - if not callable(fn): - raise TypeError("fn must be callable") - found_min, found_max = _arity(fn) - min_args = found_min if min_args is None else min_args - max_args = found_max if max_args is None else max_args - if min_args < 0 or max_args < min_args: - raise ValueError( - f"Invalid argument counts: min_args={min_args}, max_args={max_args}" - ) + min_args, max_args = check_function(name, fn, min_args, max_args) functions = _bridge_class("com.arcadedb.python.PythonFunctions") try: @@ -377,10 +396,7 @@ def __repr__(self) -> str: def register_event(database, event: str, fn: Callable, type_name: Optional[str]): """Register ``fn`` for ``event`` on the database, or on one type, and return its handle.""" - if event not in EVENTS: - raise ValueError(f"Invalid event {event!r}. Must be one of: {list(EVENTS)}") - if not callable(fn): - raise TypeError("fn must be callable") + check_event(event, fn) java_db = database.get_java_database() try: if type_name is None: diff --git a/tests/test_native_udf.py b/tests/test_native_udf.py index e5dc83b..234794a 100644 --- a/tests/test_native_udf.py +++ b/tests/test_native_udf.py @@ -9,7 +9,7 @@ def test_a_python_function_in_a_select(temp_db): - temp_db.register_function("py_double", lambda x: x * 2, 1, 1) + temp_db.register_function("py_double", lambda x: x * 2, min_args=1, max_args=1) row = temp_db.query("sql", "SELECT py_double(21) AS v").first() assert row.get("v") == 42 @@ -19,7 +19,9 @@ def test_a_python_function_reads_columns_and_returns_text(temp_db): with temp_db.transaction(): for i in range(3): temp_db.command("sql", "INSERT INTO T SET n = ?, s = ?", i, f"x{i}") - temp_db.register_function("py_tag", lambda n, s: f"{s}:{n + 1}", 2, 2) + temp_db.register_function( + "py_tag", lambda n, s: f"{s}:{n + 1}", min_args=2, max_args=2 + ) rows = temp_db.query("sql", "SELECT py_tag(n, s) AS v FROM T ORDER BY n").to_list() assert [r["v"] for r in rows] == ["x0:1", "x1:2", "x2:3"] @@ -28,7 +30,7 @@ def test_an_exception_in_the_function_fails_the_statement_with_its_text(temp_db) def boom(x): raise ValueError("no way") - temp_db.register_function("py_boom", boom, 1, 1) + temp_db.register_function("py_boom", boom, min_args=1, max_args=1) with pytest.raises(ArcadeDBError, match="no way"): temp_db.query("sql", "SELECT py_boom(1) AS v").first() @@ -41,7 +43,7 @@ def test_a_function_may_call_back_into_the_database(temp_db): def count_c(): return temp_db.query("sql", "SELECT count(*) AS c FROM C").first().get("c") - temp_db.register_function("py_count_c", count_c, 0, 0) + temp_db.register_function("py_count_c", count_c, min_args=0, max_args=0) assert temp_db.query("sql", "SELECT py_count_c() AS v").first().get("v") == 1 diff --git a/tests/test_python_callbacks.py b/tests/test_python_callbacks.py index 061293b..d3e167b 100644 --- a/tests/test_python_callbacks.py +++ b/tests/test_python_callbacks.py @@ -219,17 +219,18 @@ def test_a_numpy_array_is_a_return_value(db): strict=False, ) def test_registering_again_and_again_does_not_grow_the_python_heap(db): - """The JPype proxy class is defined once, not per registration (it leaked about 6 KB each).""" + """The JPype proxy class is defined once, not per registration (it leaked about 6 KB each); on the native backend the callback table must not keep a removed callback.""" import gc import time import tracemalloc - import jpype - def settle(): - # JPype frees a proxy's Python object once the JVM has collected the Java side - jpype.JClass("java.lang.System").gc() - time.sleep(0.3) + if arcadedb.__backend__ == "jvm": + import jpype + + # JPype frees a proxy's Python object once the JVM has collected the Java side + jpype.JClass("java.lang.System").gc() + time.sleep(0.3) gc.collect() def cycle(n): From bab98173c97c2e2dfdaf406a25737a3e18d1edfb Mon Sep 17 00:00:00 2001 From: Taewoon Kim Date: Sun, 25 Oct 2026 20:49:09 +0900 Subject: [PATCH 2/6] Native: a callback reply lives in library memory that Java frees A callback on a Java thread (an async writer) ends its Python thread state when it returns, and the thread-local buffer that held the reply went with it: the engine read freed memory ("bad JSON value at 0" for a Python function called from an async command). The reply is now allocated by the library (arcadedb_n_alloc), and Java copies the text and frees it. Co-Authored-By: Claude Sonnet 5.5 --- .../org/humemai/nativelib/NativeExtras.java | 23 ++++++++++++++++--- src/arcadedb/_native/extras.py | 9 +++++++- src/arcadedb/_native/lib.py | 12 ++++++++++ 3 files changed, 40 insertions(+), 4 deletions(-) diff --git a/native/src/main/java/org/humemai/nativelib/NativeExtras.java b/native/src/main/java/org/humemai/nativelib/NativeExtras.java index ebbddf5..df6e216 100644 --- a/native/src/main/java/org/humemai/nativelib/NativeExtras.java +++ b/native/src/main/java/org/humemai/nativelib/NativeExtras.java @@ -6,8 +6,8 @@ * Same reply convention as ArcadeNative: a malloc'd JSON text, {"ok":true,...} or {"ok":false,"error":..,"cls":..,"chain":..}. * * Upcalls. Python registers ONE ctypes function pointer with arcadedb_n_set_upcall: CCharPointer fn(long id, CCharPointer json). - * `id` names a Python-side callback, `json` is the message, the return value is a JSON reply that Python keeps alive until its - * next call on the same thread (Java copies it at once). The call is a transition-to-native C function pointer call, so the + * `id` names a Python-side callback, `json` is the message, the return value is a JSON reply in memory from arcadedb_n_alloc + * (NULL for no reply); Java copies the text and frees it. The call is a transition-to-native C function pointer call, so the * isolate can run a safepoint or a GC while the Python callback holds the GIL and calls back into the library on its thread. * ctypes takes the GIL itself (PyGILState_Ensure) when a foreign Java thread (an async writer) or the engine thread inside a * query calls the pointer. @@ -52,6 +52,7 @@ import com.arcadedb.schema.EdgeType; import com.arcadedb.schema.VertexType; import org.graalvm.nativeimage.IsolateThread; +import org.graalvm.nativeimage.UnmanagedMemory; import org.graalvm.nativeimage.c.function.CEntryPoint; import org.graalvm.nativeimage.c.function.CFunctionPointer; import org.graalvm.nativeimage.c.function.InvokeCFunctionPointer; @@ -93,10 +94,26 @@ static String upcall(final long id, final String json) { final PyCallback fn = WordFactory.pointer(raw); try (CTypeConversion.CCharPointerHolder in = CTypeConversion.toCString(json)) { final CCharPointer reply = fn.invoke(id, in.get()); - return reply.isNull() ? null : CTypeConversion.toJavaString(reply); + if (reply.isNull()) + return null; + try { + return CTypeConversion.toJavaString(reply); + } finally { + UnmanagedMemory.free(reply); + } } } + /** + * Memory for the reply of a Python callback (see upcall: Java copies the text and frees it). Python cannot keep the reply in + * its own memory: a callback on a Java thread (an async writer) ends its Python thread state when it returns, and that frees + * what the state held before Java has read the text. + */ + @CEntryPoint(name = "arcadedb_n_alloc") + public static CCharPointer alloc(final IsolateThread thread, final long size) { + return UnmanagedMemory.malloc(WordFactory.unsigned(size)); + } + /** A callback failure that must not stop an async writer: reported on stderr, not lost. */ static void swallowed(final String where, final Throwable t) { System.err.println("[arcadedb-native] " + where + " failed: " + t); diff --git a/src/arcadedb/_native/extras.py b/src/arcadedb/_native/extras.py index 96873ef..f5c02c2 100644 --- a/src/arcadedb/_native/extras.py +++ b/src/arcadedb/_native/extras.py @@ -51,7 +51,14 @@ def _dispatch(cid, raw): data = json.dumps({"error": f"{type(exc).__name__}: {exc}"}).encode("utf-8") if data is None: return None - # kept alive until this thread's next callback: Java copies the text before it returns + size = len(data) + 1 + pointer = lib.alloc(size) + if pointer: + # library memory: Java copies the text and frees it (on a Java thread this callback's Python thread state, and + # everything a thread-local held, is gone when it returns) + ctypes.memmove(pointer, data + b"\0", size) + return pointer + # a library older than arcadedb_n_alloc: kept alive until this thread's next callback buf = ctypes.create_string_buffer(data) _reply.buf = buf return ctypes.addressof(buf) diff --git a/src/arcadedb/_native/lib.py b/src/arcadedb/_native/lib.py index d0c05d5..c02616a 100644 --- a/src/arcadedb/_native/lib.py +++ b/src/arcadedb/_native/lib.py @@ -179,6 +179,7 @@ class _Params(ctypes.Structure): "arcadedb_n_import": [P, L, S], "arcadedb_n_export": [P, L, S, S, I, S], "arcadedb_n_cb": [P, S, S], + "arcadedb_n_alloc": [P, L], "arcadedb_n_set_upcall": [P, P], "arcadedb_free": [P, P], } @@ -277,6 +278,17 @@ def call_bin(name: str, *args): return raw.decode("utf-8"), floats +def alloc(size: int): + """Memory from the library's allocator, or None when the library is older than ``arcadedb_n_alloc``. + + The reply of a callback goes through it: Java copies the text and frees the memory, so the reply outlives the callback's + Python thread state (a callback on an async writer thread ends that state as it returns). + """ + lib = _lib or _load() + fn = getattr(lib, "arcadedb_n_alloc", None) + return None if fn is None else fn(_thread_handle(), size) + + def call_void(name: str, *args) -> None: """Run an entry point that returns nothing (``arcadedb_n_set_upcall``).""" lib = _lib or _load() From bae8dfc5cfde0d790ac66cc8aa8765e0131e84a7 Mon Sep 17 00:00:00 2001 From: Taewoon Kim Date: Sun, 25 Oct 2026 21:13:36 +0900 Subject: [PATCH 3/6] Native: a vetoed save is not an error; an after_delete event of an unloaded record reports only its identity save() of a new record whose before_create callback vetoed it returns the unsaved record, as on the JPype backend, instead of raising "the statement returned no record". The after_delete event of a record that was not loaded (a delete by RID) found no record to read the properties from and failed the delete; it sends the identity only now (Java change, in the next native image). Co-Authored-By: Claude Sonnet 5.5 --- .../org/humemai/nativelib/NativeExtras.java | 17 +++++++++++------ src/arcadedb/_native/extras.py | 13 ++++++++++++- src/arcadedb/_native/records.py | 5 +++++ 3 files changed, 28 insertions(+), 7 deletions(-) diff --git a/native/src/main/java/org/humemai/nativelib/NativeExtras.java b/native/src/main/java/org/humemai/nativelib/NativeExtras.java index df6e216..aa54e4b 100644 --- a/native/src/main/java/org/humemai/nativelib/NativeExtras.java +++ b/native/src/main/java/org/humemai/nativelib/NativeExtras.java @@ -744,13 +744,18 @@ static void appendRecord(final StringBuilder sb, final Document doc) { sb.append("\"@type\":"); NativeCodec.quote(sb, doc.getTypeName()); sb.append(",\"@cat\":\"").append(doc instanceof Vertex ? 'v' : doc instanceof Edge ? 'e' : 'd').append('"'); - for (final String name : doc.getPropertyNames()) { - sb.append(','); - NativeCodec.quote(sb, name); - sb.append(':'); - NativeCodec.append(sb, doc.get(name)); + final StringBuilder props = new StringBuilder(); + try { + for (final String name : doc.getPropertyNames()) { + props.append(','); + NativeCodec.quote(props, name); + props.append(':'); + NativeCodec.append(props, doc.get(name)); + } + } catch (final com.arcadedb.exception.RecordNotFoundException e) { + props.setLength(0); // an after_delete event of a record that was not loaded: it is gone, only its identity is left to report } - sb.append('}'); + sb.append(props).append('}'); } /** diff --git a/src/arcadedb/_native/extras.py b/src/arcadedb/_native/extras.py index f5c02c2..a580c99 100644 --- a/src/arcadedb/_native/extras.py +++ b/src/arcadedb/_native/extras.py @@ -868,6 +868,14 @@ def handler(message): return handler +_veto = threading.local() + + +def veto_count() -> int: + """How many ``before_*`` callbacks vetoed a change on this thread so far (``save`` tells a veto from a failure by it).""" + return getattr(_veto, "count", 0) + + def _event_handler(event: str, fn: Callable, database_ref) -> Callable: is_before = event.startswith("before_") @@ -875,7 +883,10 @@ def handler(message): from .records import _wrap_dict result = fn(_wrap_dict(database_ref(), message["rec"])) - return {"v": not (is_before and result is False)} + vetoed = is_before and result is False + if vetoed: + _veto.count = veto_count() + 1 + return {"v": not vetoed} return handler diff --git a/src/arcadedb/_native/records.py b/src/arcadedb/_native/records.py index 7f03d2f..dcdbd1b 100644 --- a/src/arcadedb/_native/records.py +++ b/src/arcadedb/_native/records.py @@ -99,9 +99,14 @@ def set(self, name: str, value: Any): def save(self): db = self._db if self._rid is None: + from .extras import veto_count + + vetoes = veto_count() row = db.command( "sql", self._CREATE.format(t=_q(self._type)), {"row": self._props} ).first() + if row is None and veto_count() != vetoes: + return self # a before_create callback vetoed it: nothing is written, no error (as on the JPype backend) self._adopt(row) else: db.command( From fbd8664fa74d3527b51e0f0bbf76039977377e2a Mon Sep 17 00:00:00 2001 From: Taewoon Kim Date: Sun, 25 Oct 2026 21:22:10 +0900 Subject: [PATCH 4/6] Native: the reply of a callback goes through an out slot, never a return value A Ctrl-C raised as Python enters the callback (before its try) left the C return value unset, and the engine took whatever the return register held: the pointer of an earlier reply that Java had freed already. Java then freed it again and the process aborted in free(). The callback is void now and writes its reply to a slot that Java zeroes before the call, so a callback that ends early leaves NULL; the failed statement becomes a KeyboardInterrupt. Python writes only library memory into the slot (the fallback to a Python-owned buffer is gone), and the unraisablehook is not used. A test runs the SIGINT loop in a fresh process and fails on a nonzero exit. Co-Authored-By: Claude Sonnet 5.5 --- .../org/humemai/nativelib/NativeExtras.java | 21 +++++--- src/arcadedb/_native/extras.py | 49 +++++------------ src/arcadedb/_native/interrupts.py | 12 ++++- src/arcadedb/_native/lib.py | 7 ++- tests/test_python_function_interrupt.py | 54 +++++++++++++++++++ 5 files changed, 96 insertions(+), 47 deletions(-) diff --git a/native/src/main/java/org/humemai/nativelib/NativeExtras.java b/native/src/main/java/org/humemai/nativelib/NativeExtras.java index aa54e4b..9903789 100644 --- a/native/src/main/java/org/humemai/nativelib/NativeExtras.java +++ b/native/src/main/java/org/humemai/nativelib/NativeExtras.java @@ -5,9 +5,9 @@ * database settings, import and export, and the Java->Python upcall (async callbacks, Python functions in SQL). * Same reply convention as ArcadeNative: a malloc'd JSON text, {"ok":true,...} or {"ok":false,"error":..,"cls":..,"chain":..}. * - * Upcalls. Python registers ONE ctypes function pointer with arcadedb_n_set_upcall: CCharPointer fn(long id, CCharPointer json). - * `id` names a Python-side callback, `json` is the message, the return value is a JSON reply in memory from arcadedb_n_alloc - * (NULL for no reply); Java copies the text and frees it. The call is a transition-to-native C function pointer call, so the + * Upcalls. Python registers ONE ctypes function pointer with arcadedb_n_set_upcall: void fn(long id, CCharPointer json, void **out). + * `id` names a Python-side callback, `json` is the message, the JSON reply is written to *out as memory from arcadedb_n_alloc + * (left NULL for no reply); Java copies the text and frees it. The call is a transition-to-native C function pointer call, so the * isolate can run a safepoint or a GC while the Python callback holds the GIL and calls back into the library on its thread. * ctypes takes the GIL itself (PyGILState_Ensure) when a foreign Java thread (an async writer) or the engine thread inside a * query calls the pointer. @@ -58,6 +58,7 @@ import org.graalvm.nativeimage.c.function.InvokeCFunctionPointer; import org.graalvm.nativeimage.c.type.CCharPointer; import org.graalvm.nativeimage.c.type.CTypeConversion; +import org.graalvm.word.Pointer; import org.graalvm.word.WordFactory; import java.util.ArrayList; @@ -76,7 +77,7 @@ private NativeExtras() { interface PyCallback extends CFunctionPointer { @InvokeCFunctionPointer - CCharPointer invoke(long id, CCharPointer json); + void invoke(long id, CCharPointer json, Pointer out); } private static volatile long upcallPointer; @@ -92,15 +93,23 @@ static String upcall(final long id, final String json) { if (raw == 0) throw new IllegalStateException("no Python callback is registered (arcadedb_n_set_upcall)"); final PyCallback fn = WordFactory.pointer(raw); + // The reply comes back through *out, which is zeroed here, never through the return value of the call: a callback that + // fails before it writes (a Ctrl-C raised as Python enters it) leaves a NULL, where a C return value would be whatever the + // return register held. + final Pointer out = UnmanagedMemory.malloc(8); + out.writeLong(0, 0L); try (CTypeConversion.CCharPointerHolder in = CTypeConversion.toCString(json)) { - final CCharPointer reply = fn.invoke(id, in.get()); + fn.invoke(id, in.get(), out); + final CCharPointer reply = out.readWord(0); if (reply.isNull()) return null; try { return CTypeConversion.toJavaString(reply); } finally { - UnmanagedMemory.free(reply); + UnmanagedMemory.free(reply); // memory of arcadedb_n_alloc, the only kind Python writes into *out } + } finally { + UnmanagedMemory.free(out); } } diff --git a/src/arcadedb/_native/extras.py b/src/arcadedb/_native/extras.py index a580c99..072184e 100644 --- a/src/arcadedb/_native/extras.py +++ b/src/arcadedb/_native/extras.py @@ -10,7 +10,6 @@ import ctypes import itertools import json -import sys import threading import weakref from collections.abc import Mapping @@ -25,13 +24,13 @@ # -------------------------------------------------------------------------------------------------------- upcall +# (Java long id, char *json, void **out): the reply goes through *out, not through a return value (see _dispatch) _UPCALL_TYPE = ctypes.CFUNCTYPE( - ctypes.c_void_p, ctypes.c_longlong, ctypes.c_char_p -) # (Java long id, char *json) + None, ctypes.c_longlong, ctypes.c_char_p, ctypes.c_void_p +) _callbacks: dict = {} _callback_ids = itertools.count(1) _callbacks_lock = threading.Lock() -_reply = threading.local() _upcall_ref = None @@ -40,7 +39,15 @@ def _enc(value) -> str: return json.dumps(_prep(value), default=_default) -def _dispatch(cid, raw): +def _dispatch(cid, raw, out): + """Run callback ``cid`` with the message ``raw`` and write the JSON reply to ``*out`` (Java zeroed it before the call). + + The reply never travels as a return value. A C callback that returns without setting its return value hands the engine + whatever the return register held, here the pointer of an earlier reply, which Java had already freed: an abort in + ``free()``. That is what happens when a Ctrl-C is raised as this function is entered, before its ``try``. Through ``*out`` + the engine sees NULL then, fails the statement, and the error becomes a KeyboardInterrupt (``interrupts.convert``). + The memory comes from the library (``arcadedb_n_alloc``): Java copies the text and frees it, so it must be nothing else. + """ try: fn = _callbacks.get(cid) message = loads(raw.decode("utf-8")) if raw else {} @@ -50,46 +57,18 @@ def _dispatch(cid, raw): interrupts.note(exc) data = json.dumps({"error": f"{type(exc).__name__}: {exc}"}).encode("utf-8") if data is None: - return None + return size = len(data) + 1 pointer = lib.alloc(size) if pointer: - # library memory: Java copies the text and frees it (on a Java thread this callback's Python thread state, and - # everything a thread-local held, is gone when it returns) ctypes.memmove(pointer, data + b"\0", size) - return pointer - # a library older than arcadedb_n_alloc: kept alive until this thread's next callback - buf = ctypes.create_string_buffer(data) - _reply.buf = buf - return ctypes.addressof(buf) - - -def _install_unraisable_hook() -> None: - """Keep a Ctrl-C that arrives as the callback is entered. - - CPython runs a pending signal handler at the first bytecode of a function, so a KeyboardInterrupt can be raised before - ``_dispatch`` reaches its ``try``: ctypes then reports it as unraisable and returns NULL to the engine. The hook records it - as the interrupt of this thread (the engine fails the statement on a NULL reply, and the error becomes the - KeyboardInterrupt); every other unraisable exception goes to the hook that was there. - """ - previous = sys.unraisablehook - - def hook(unraisable): - if unraisable.object is _dispatch and isinstance( - unraisable.exc_value, KeyboardInterrupt - ): - interrupts.note(unraisable.exc_value) - else: - previous(unraisable) - - sys.unraisablehook = hook + ctypes.c_void_p.from_address(out).value = pointer def _ensure_upcall() -> None: global _upcall_ref if _upcall_ref is not None: return - _install_unraisable_hook() lib.call_void( "arcadedb_n_set_upcall", ctypes.cast(_make_pointer(), ctypes.c_void_p) ) diff --git a/src/arcadedb/_native/interrupts.py b/src/arcadedb/_native/interrupts.py index c3bfcaa..cfe990f 100644 --- a/src/arcadedb/_native/interrupts.py +++ b/src/arcadedb/_native/interrupts.py @@ -23,11 +23,19 @@ def clear() -> None: _PENDING.interrupt = None +# What the engine says when a callback returned no reply (NativeExtras.answer): the callback ended before it could write one. +# That happens when a Ctrl-C is raised as Python enters the callback, before its code (and its try) runs. +_NO_REPLY = "was interrupted before Python returned a value" + + def convert(error: BaseException) -> BaseException: - """The ``KeyboardInterrupt`` to raise in place of ``error`` if a callback left one on this thread, else ``error``.""" + """The ``KeyboardInterrupt`` to raise in place of ``error``: one a callback left on this thread, or the Ctrl-C that + ended a callback before it could reply. Else ``error`` itself.""" interrupt = getattr(_PENDING, "interrupt", None) if interrupt is None: - return error + if _NO_REPLY not in str(error): + return error + interrupt = KeyboardInterrupt("interrupted while a Python function was called") _PENDING.interrupt = None interrupt.__cause__ = error return interrupt diff --git a/src/arcadedb/_native/lib.py b/src/arcadedb/_native/lib.py index c02616a..95cc7a3 100644 --- a/src/arcadedb/_native/lib.py +++ b/src/arcadedb/_native/lib.py @@ -279,14 +279,13 @@ def call_bin(name: str, *args): def alloc(size: int): - """Memory from the library's allocator, or None when the library is older than ``arcadedb_n_alloc``. + """Memory from the library's allocator (``arcadedb_n_alloc``), for the reply of a callback. - The reply of a callback goes through it: Java copies the text and frees the memory, so the reply outlives the callback's + Java copies the text and frees the memory, so it must come from here and from nowhere else; it also outlives the callback's Python thread state (a callback on an async writer thread ends that state as it returns). """ lib = _lib or _load() - fn = getattr(lib, "arcadedb_n_alloc", None) - return None if fn is None else fn(_thread_handle(), size) + return lib.arcadedb_n_alloc(_thread_handle(), size) def call_void(name: str, *args) -> None: diff --git a/tests/test_python_function_interrupt.py b/tests/test_python_function_interrupt.py index c2a3332..0bcc87b 100644 --- a/tests/test_python_function_interrupt.py +++ b/tests/test_python_function_interrupt.py @@ -11,6 +11,8 @@ import os import signal +import subprocess +import sys import threading import time @@ -82,3 +84,55 @@ def test_a_statement_that_fails_otherwise_is_still_an_arcadedb_error(db): with pytest.raises(arcadedb.ArcadeDBError) as raised: db.query("sql", "SELECT FROM NoSuchType") assert not isinstance(raised.value, KeyboardInterrupt) + + +_SIGINT_LOOP = r""" +import os, signal, tempfile, threading, time +import numpy as np +import arcadedb + +with tempfile.TemporaryDirectory() as directory: + db = arcadedb.create_database(os.path.join(directory, "db")) + db.command("sql", "CREATE DOCUMENT TYPE T") + db.insert_columns("T", {"n": np.arange(150_000, dtype=np.int64)}) + db.register_function("plus_one", lambda x: x + 1) + interrupted = 0 + for delay in (0.05, 0.15, 0.25, 0.35, 0.45, 0.55): + timer = threading.Timer(delay, os.kill, (os.getpid(), signal.SIGINT)) + timer.start() + try: + db.query("sql", "SELECT sum(plus_one(n)) AS s FROM T").first() + done = True + except KeyboardInterrupt: + done = False + interrupted += 1 + finally: + timer.cancel() + timer.join() + if done: + try: + time.sleep(0.05) + except KeyboardInterrupt: + interrupted += 1 + assert db.count_type("T") == 150_000 + db.close() + print("interrupted", interrupted) +""" + + +@pytest.mark.skipif( + sys.platform == "win32", reason="a signal cannot be sent to the own process here" +) +def test_sigint_loop_in_a_fresh_process_ends_cleanly(): + """The same loop in its own process. A Ctrl-C raised as the Python callback is entered used to leave the engine a stale + reply pointer, and the process aborted in ``free()`` (nonzero exit); pytest's own hooks hid it inside a test run. + """ + result = subprocess.run( + [sys.executable, "-c", _SIGINT_LOOP], + capture_output=True, + text=True, + timeout=600, + check=False, + ) + assert result.returncode == 0, result.stderr[-2000:] + assert int(result.stdout.split()[-1]) >= 1, result.stdout From 702e49c39602fe1a2c1fdac45b32bad21cded54c Mon Sep 17 00:00:00 2001 From: Taewoon Kim Date: Sun, 25 Oct 2026 21:28:16 +0900 Subject: [PATCH 5/6] Native: restore the schema property; the SIGINT subprocess test installs the default handler NativeDatabase.on() took the @property line of schema with it when register_function was replaced; db.schema was a bound method again. The fresh-process SIGINT test sets the default Ctrl-C handler in the child: a parent started as a background job ignores SIGINT, and Python then installs no handler. Co-Authored-By: Claude Sonnet 5.5 --- src/arcadedb/_native/database.py | 1 + tests/test_python_function_interrupt.py | 2 ++ 2 files changed, 3 insertions(+) diff --git a/src/arcadedb/_native/database.py b/src/arcadedb/_native/database.py index 9646cbb..c8695aa 100644 --- a/src/arcadedb/_native/database.py +++ b/src/arcadedb/_native/database.py @@ -916,6 +916,7 @@ def on(self, event, fn, type_name=None): self._check_not_closed() return self._registrations().on(event, fn, type_name) + @property def schema(self): """DDL helpers over SQL statements (the DSL-first subset); everything else is not on native yet.""" self._check_not_closed() diff --git a/tests/test_python_function_interrupt.py b/tests/test_python_function_interrupt.py index 0bcc87b..4a0d031 100644 --- a/tests/test_python_function_interrupt.py +++ b/tests/test_python_function_interrupt.py @@ -91,6 +91,8 @@ def test_a_statement_that_fails_otherwise_is_still_an_arcadedb_error(db): import numpy as np import arcadedb +# a parent started as a background job ignores SIGINT, and a child inherits that: Python then installs no handler +signal.signal(signal.SIGINT, signal.default_int_handler) with tempfile.TemporaryDirectory() as directory: db = arcadedb.create_database(os.path.join(directory, "db")) db.command("sql", "CREATE DOCUMENT TYPE T") From 648c62b87309a46544eab42bbd3cc2a646ffba00 Mon Sep 17 00:00:00 2001 From: Taewoon Kim Date: Sun, 25 Oct 2026 21:29:57 +0900 Subject: [PATCH 6/6] Tests: nosec markers for the fresh-process SIGINT test's subprocess Co-Authored-By: Claude Opus 5.5 --- tests/test_python_function_interrupt.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/test_python_function_interrupt.py b/tests/test_python_function_interrupt.py index 4a0d031..0ca4ddd 100644 --- a/tests/test_python_function_interrupt.py +++ b/tests/test_python_function_interrupt.py @@ -11,7 +11,7 @@ import os import signal -import subprocess +import subprocess # nosec B404 - test-controlled child process import sys import threading import time @@ -129,7 +129,7 @@ def test_sigint_loop_in_a_fresh_process_ends_cleanly(): """The same loop in its own process. A Ctrl-C raised as the Python callback is entered used to leave the engine a stale reply pointer, and the process aborted in ``free()`` (nonzero exit); pytest's own hooks hid it inside a test run. """ - result = subprocess.run( + result = subprocess.run( # nosec B603 - fixed argv, no shell, test-owned code string [sys.executable, "-c", _SIGINT_LOOP], capture_output=True, text=True,