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..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 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 + * 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. @@ -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,12 +48,17 @@ 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.UnmanagedMemory; import org.graalvm.nativeimage.c.function.CEntryPoint; import org.graalvm.nativeimage.c.function.CFunctionPointer; 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; @@ -59,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; @@ -75,12 +93,36 @@ 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()); - return reply.isNull() ? null : CTypeConversion.toJavaString(reply); + 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); // memory of arcadedb_n_alloc, the only kind Python writes into *out + } + } finally { + UnmanagedMemory.free(out); } } + /** + * 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); @@ -240,7 +282,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 +429,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 +596,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 +708,114 @@ 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('"'); + 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(props).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 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"); + 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); + } + } + + private String arity() { + if (min == max) + return String.valueOf(min); + return max == Integer.MAX_VALUE ? "at least " + min : min + " to " + max; } @Override @@ -692,17 +828,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); + } } } - @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) { + 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_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..c8695aa 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,11 +882,39 @@ 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): diff --git a/src/arcadedb/_native/extras.py b/src/arcadedb/_native/extras.py index 69c6789..072184e 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)``, @@ -11,23 +11,26 @@ import itertools import json 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 +# (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 @@ -36,20 +39,30 @@ 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 {} 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 - # kept alive until this thread's next callback: Java copies the text before it returns - buf = ctypes.create_string_buffer(data) - _reply.buf = buf - return ctypes.addressof(buf) + return + size = len(data) + 1 + pointer = lib.alloc(size) + if pointer: + ctypes.memmove(pointer, data + b"\0", size) + ctypes.c_void_p.from_address(out).value = pointer def _ensure_upcall() -> None: @@ -67,6 +80,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 +809,175 @@ 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 + + +_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_") + + def handler(message): + from .records import _wrap_dict + + result = fn(_wrap_dict(database_ref(), message["rec"])) + vetoed = is_before and result is False + if vetoed: + _veto.count = veto_count() + 1 + return {"v": not vetoed} + + 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..cfe990f --- /dev/null +++ b/src/arcadedb/_native/interrupts.py @@ -0,0 +1,41 @@ +"""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 + + +# 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``: 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: + 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 b2f16ac..95cc7a3 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,8 @@ 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_alloc": [P, L], "arcadedb_n_set_upcall": [P, P], "arcadedb_free": [P, P], } @@ -222,6 +225,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: @@ -268,6 +278,16 @@ def call_bin(name: str, *args): return raw.decode("utf-8"), floats +def alloc(size: int): + """Memory from the library's allocator (``arcadedb_n_alloc``), for the reply of a callback. + + 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() + return lib.arcadedb_n_alloc(_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() 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( 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): diff --git a/tests/test_python_function_interrupt.py b/tests/test_python_function_interrupt.py index c2a3332..0ca4ddd 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 # nosec B404 - test-controlled child process +import sys import threading import time @@ -82,3 +84,57 @@ 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 + +# 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") + 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( # nosec B603 - fixed argv, no shell, test-owned code string + [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