From c999c6003eb4b59ac72561934b17679faabd093b Mon Sep 17 00:00:00 2001 From: 1ndahous3 <9205230+1ndahous3@users.noreply.github.com> Date: Thu, 15 Oct 2026 14:27:15 +0300 Subject: [PATCH 1/5] Release SQLite callbacks and expanded trace SQL safely Assisted-by: Codex:gpt-6 --- crates/stdlib/src/_sqlite3.rs | 256 +++++++++++++++++++------- extra_tests/snippets/stdlib_sqlite.py | 74 ++++++++ 2 files changed, 259 insertions(+), 71 deletions(-) diff --git a/crates/stdlib/src/_sqlite3.rs b/crates/stdlib/src/_sqlite3.rs index c03cfe73e5b..c7d1da51cd5 100644 --- a/crates/stdlib/src/_sqlite3.rs +++ b/crates/stdlib/src/_sqlite3.rs @@ -34,10 +34,10 @@ mod _sqlite3 { sqlite3_create_function_v2, sqlite3_create_window_function, sqlite3_data_count, sqlite3_db_config, sqlite3_db_handle, sqlite3_db_mutex, sqlite3_errcode, sqlite3_errmsg, sqlite3_exec, sqlite3_expanded_sql, sqlite3_extended_errcode, sqlite3_finalize, - sqlite3_get_autocommit, sqlite3_interrupt, sqlite3_last_insert_rowid, sqlite3_libversion, - sqlite3_limit, sqlite3_mutex_enter, sqlite3_mutex_leave, sqlite3_mutex_try, - sqlite3_open_v2, sqlite3_prepare_v2, sqlite3_progress_handler, sqlite3_reset, - sqlite3_result_blob, sqlite3_result_double, sqlite3_result_error, + sqlite3_free, sqlite3_get_autocommit, sqlite3_interrupt, sqlite3_last_insert_rowid, + sqlite3_libversion, sqlite3_limit, sqlite3_mutex_enter, sqlite3_mutex_leave, + sqlite3_mutex_try, sqlite3_open_v2, sqlite3_prepare_v2, sqlite3_progress_handler, + sqlite3_reset, sqlite3_result_blob, sqlite3_result_double, sqlite3_result_error, sqlite3_result_error_nomem, sqlite3_result_error_toobig, sqlite3_result_int64, sqlite3_result_null, sqlite3_result_text, sqlite3_set_authorizer, sqlite3_sleep, sqlite3_step, sqlite3_stmt, sqlite3_stmt_busy, sqlite3_stmt_readonly, sqlite3_threadsafe, @@ -478,6 +478,15 @@ mod _sqlite3 { rustpython_vm::vm::thread::with_vm(obj, |vm| f(obj, vm)) } + // The SQLite mutex keeps the registration alive until this reference is acquired. + unsafe fn registered(data: *mut c_void) -> PyRc { + let data = data.cast::(); + unsafe { + PyRc::increment_strong_count(data); + PyRc::from_raw(data) + } + } + fn with_context(&self, context: SqliteContext, f: impl Fn(&PyObject, &VirtualMachine)) { if self.with_vm(f).is_none() { unsafe { @@ -653,7 +662,7 @@ mod _sqlite3 { db_name: *const libc::c_char, access: *const libc::c_char, ) -> c_int { - let data = unsafe { &*data.cast::() }; + let data = unsafe { Self::registered(data) }; data.with_vm(|callable, vm| { let f = || -> PyResult { let arg1 = ptr_to_str_or_none(arg1, vm)?; @@ -679,11 +688,18 @@ mod _sqlite3 { stmt: *mut c_void, sql: *mut c_void, ) -> c_int { - let data = unsafe { &*data.cast::() }; + let data = unsafe { Self::registered(data) }; data.with_vm(|callable, vm| { - let expanded = unsafe { sqlite3_expanded_sql(stmt.cast()) }; let f = || -> PyResult<()> { - let stmt = ptr_to_str(expanded, vm).or_else(|_| ptr_to_str(sql.cast(), vm))?; + let stmt = { + let expanded = scopeguard::guard( + unsafe { sqlite3_expanded_sql(stmt.cast()) }, + |sql| unsafe { sqlite3_free(sql.cast()) }, + ); + let text = + ptr_to_str(*expanded, vm).or_else(|_| ptr_to_str(sql.cast(), vm))?; + vm.ctx.new_str(text) + }; callable.call((stmt,), vm)?; Ok(()) }; @@ -694,7 +710,7 @@ mod _sqlite3 { } unsafe extern "C" fn progress_callback(data: *mut c_void) -> c_int { - let data = unsafe { &*data.cast::() }; + let data = unsafe { Self::registered(data) }; data.with_vm(|callable, vm| { if let Ok(val) = callable.call((), vm) && let Ok(val) = val.is_true(vm) @@ -787,9 +803,10 @@ mod _sqlite3 { impl Drop for CallbackData { fn drop(&mut self) { - unsafe { - let _ = PyObjectRef::from_raw(self.obj); - }; + let obj = unsafe { PyObjectRef::from_raw(self.obj) }; + // The last active callback may outlive its registration. Release the + // Python reference after the outer operation's native/Rust locks. + try_defer_drop(move || drop(obj)); } } @@ -1036,6 +1053,12 @@ mod _sqlite3 { type Args = FuncArgs; fn call(zelf: &Py, args: Self::Args, vm: &VirtualMachine) -> PyResult { + with_deferred_drops(|| Self::call_inner(zelf, args, vm)) + } + } + + impl Connection { + fn call_inner(zelf: &Py, args: FuncArgs, vm: &VirtualMachine) -> PyResult { let _ = zelf.db_lock(vm)?; let (sql,): (PyUtf8StrRef,) = args.bind(vm)?; @@ -1158,10 +1181,8 @@ mod _sqlite3 { }; drop(statements); let rollback = *self.autocommit.lock() == AutocommitMode::Disabled; - if let Some(db) = db - && rollback - { - db.rollback_on_close(vm)?; + if let Some(db) = db { + db.close_callbacks(rollback, vm)?; } Ok(()) } @@ -1683,22 +1704,7 @@ mod _sqlite3 { callable: PyObjectRef, vm: &VirtualMachine, ) -> PyResult<()> { - let db = zelf.db_lock(vm)?; - let Some(data) = CallbackData::new(callable, vm) else { - unsafe { sqlite3_set_authorizer(db.db, None, null_mut()) }; - return Ok(()); - }; - - let ret = db.call(|| unsafe { - sqlite3_set_authorizer( - db.db, - Some(CallbackData::authorizer_callback), - Box::into_raw(Box::new(data)).cast(), - ) - }); - db.check(ret, vm).map_err(|_| { - new_operational_error(vm, "Error setting authorizer callback".to_owned()) - }) + zelf.set_callback(callable, CallbackKind::Authorizer, vm) } #[pymethod] @@ -1707,22 +1713,7 @@ mod _sqlite3 { callable: PyObjectRef, vm: &VirtualMachine, ) -> PyResult<()> { - let db = zelf.db_lock(vm)?; - let Some(data) = CallbackData::new(callable, vm) else { - unsafe { sqlite3_trace_v2(db.db, SQLITE_TRACE_STMT, None, null_mut()) }; - return Ok(()); - }; - - let ret = db.call(|| unsafe { - sqlite3_trace_v2( - db.db, - SQLITE_TRACE_STMT, - Some(CallbackData::trace_callback), - Box::into_raw(Box::new(data)).cast(), - ) - }); - - db.check(ret, vm) + zelf.set_callback(callable, CallbackKind::Trace, vm) } #[pymethod] @@ -1732,22 +1723,53 @@ mod _sqlite3 { n: c_int, vm: &VirtualMachine, ) -> PyResult<()> { - let db = zelf.db_lock(vm)?; - let Some(data) = CallbackData::new(callable, vm) else { - unsafe { sqlite3_progress_handler(db.db, n, None, null_mut()) }; - return Ok(()); - }; - - unsafe { - sqlite3_progress_handler( - db.db, - n, - Some(CallbackData::progress_callback), - Box::into_raw(Box::new(data)).cast(), - ) - }; + zelf.set_callback(callable, CallbackKind::Progress(n), vm) + } - Ok(()) + fn set_callback( + &self, + callable: PyObjectRef, + kind: CallbackKind, + vm: &VirtualMachine, + ) -> PyResult<()> { + with_deferred_drops(|| { + let db = self.db_lock(vm)?; + let mut data = CallbackData::new(callable, vm).map(PyRc::new); + let pointer = data + .as_ref() + .map_or(null_mut(), |data| PyRc::as_ptr(data).cast_mut()); + let (closed, result, previous) = { + let _guard = db.lock(); + let mut callbacks = db.callbacks.lock(); + if callbacks.closed { + (true, Ok(SQLITE_OK), None) + } else { + let result = db.call(|| kind.register(db.raw, pointer)); + let previous = if result.is_ok() { + core::mem::replace(kind.slot(&mut callbacks), data.take()) + } else { + None + }; + (false, result, previous) + } + }; + // Both the native mutex and the slots mutex must be released + // before a displaced callable can run __del__ or weakref callbacks. + drop((previous, data)); + if closed { + return Err(new_programming_error( + vm, + "Cannot operate on a closed database.".to_owned(), + )); + } + db.check(result, vm).map_err(|error| { + if matches!(kind, CallbackKind::Authorizer) { + new_operational_error(vm, "Error setting authorizer callback".to_owned()) + } else { + error + } + }) + }) } #[pymethod] @@ -3371,17 +3393,72 @@ mod _sqlite3 { struct Sqlite { raw: SqliteRaw, pending: PyMutex, + callbacks: PyMutex, + } + + #[derive(Default)] + struct CallbackSlots { + trace: Option>, + progress: Option>, + authorizer: Option>, + closed: bool, + } + + #[derive(Clone, Copy)] + enum CallbackKind { + Trace, + Progress(c_int), + Authorizer, + } + + impl CallbackKind { + fn slot(self, slots: &mut CallbackSlots) -> &mut Option> { + match self { + Self::Trace => &mut slots.trace, + Self::Progress(_) => &mut slots.progress, + Self::Authorizer => &mut slots.authorizer, + } + } + + // The caller holds the native mutex and retains data until unregistration. + fn register(self, db: SqliteRaw, data: *mut CallbackData) -> c_int { + unsafe { + match self { + Self::Trace => sqlite3_trace_v2( + db.db, + SQLITE_TRACE_STMT, + (!data.is_null()).then_some(CallbackData::trace_callback), + data.cast(), + ), + Self::Progress(n) => { + sqlite3_progress_handler( + db.db, + n, + (!data.is_null()).then_some(CallbackData::progress_callback), + data.cast(), + ); + SQLITE_OK + } + Self::Authorizer => sqlite3_set_authorizer( + db.db, + (!data.is_null()).then_some(CallbackData::authorizer_callback), + data.cast(), + ), + } + } + } } #[derive(Default)] struct SqliteCleanup { statements: Vec, rollback: bool, + clear_callbacks: bool, } impl SqliteCleanup { fn is_empty(&self) -> bool { - self.statements.is_empty() && !self.rollback + self.statements.is_empty() && !self.rollback && !self.clear_callbacks } } @@ -3390,6 +3467,7 @@ mod _sqlite3 { Self { raw, pending: PyMutex::new(SqliteCleanup::default()), + callbacks: PyMutex::new(CallbackSlots::default()), } } } @@ -3431,7 +3509,9 @@ mod _sqlite3 { for statement in pending.statements { unsafe { sqlite3_finalize(statement.st) }; } + let callbacks = pending.clear_callbacks.then(|| self.clear_callbacks()); drop(guard); + drop(callbacks); // An enqueue after this check can acquire the native mutex and // drain itself; one made while it was locked must be retried. if self.pending.lock().is_empty() { @@ -3441,17 +3521,44 @@ mod _sqlite3 { }); } - fn rollback_on_close(&self, vm: &VirtualMachine) -> PyResult<()> { + fn close_callbacks(&self, rollback: bool, vm: &VirtualMachine) -> PyResult<()> { + self.callbacks.lock().closed = true; if let Some(_guard) = self.raw.try_lock() { - if !self.is_autocommit() { - self._exec(c"ROLLBACK", vm)?; - } + let result = if rollback && !self.is_autocommit() { + self._exec(c"ROLLBACK", vm) + } else { + Ok(()) + }; + self.pending.lock().clear_callbacks = true; + self.flush_cleanup(); + result } else { - self.pending.lock().rollback = true; + { + let mut pending = self.pending.lock(); + pending.rollback |= rollback; + pending.clear_callbacks = true; + } // Retry after enqueueing, in case the active call just ended. self.flush_cleanup(); + Ok(()) } - Ok(()) + } + + // The native mutex is held. SQLite must stop using the pointers first. + fn clear_callbacks(&self) -> [Option>; 3] { + for kind in [ + CallbackKind::Trace, + CallbackKind::Progress(0), + CallbackKind::Authorizer, + ] { + kind.register(self.raw, null_mut()); + } + let mut slots = self.callbacks.lock(); + [ + slots.trace.take(), + slots.progress.take(), + slots.authorizer.take(), + ] } } @@ -3523,7 +3630,14 @@ mod _sqlite3 { // queued by their destructors before closing the database. self.drain_cleanup(true); // Use sqlite3_close_v2 for safe closing even with active statements - with_deferred_drops(|| unsafe { sqlite3_close_v2(self.raw.db) }); + with_deferred_drops(|| { + let callbacks = { + let _guard = self.raw.lock(); + self.clear_callbacks() + }; + drop(callbacks); + unsafe { sqlite3_close_v2(self.raw.db) }; + }); } } diff --git a/extra_tests/snippets/stdlib_sqlite.py b/extra_tests/snippets/stdlib_sqlite.py index 9cf6e4bd3fd..1a3b517ea26 100644 --- a/extra_tests/snippets/stdlib_sqlite.py +++ b/extra_tests/snippets/stdlib_sqlite.py @@ -809,3 +809,77 @@ def run(setter): script_stops_after_close() concurrent_sqlite_results() concurrent_sqlite_transactions() + + +def callback_registration_lifetimes(): + class Callback: + def __init__(self, connection): + self.connection = connection + + def __call__(self, *args): + return sqlite.SQLITE_OK + + for method, args in ( + ("set_trace_callback", ()), + ("set_authorizer", ()), + ("set_progress_handler", (1,)), + ("set_progress_handler", (0,)), + ): + for action in ("disable", "replace", "close", "reinit"): + connection = sqlite.connect(":memory:") + cursor = connection.execute("SELECT 1 UNION ALL SELECT 2") + register = getattr(connection, method) + callback = Callback(connection) + reference = weakref.ref(callback) + register(callback, *args) + del callback + assert reference() is not None + if action == "disable": + register(None, *args) + elif action == "replace": + register(lambda *args: 0, *args) + elif action == "close": + connection.close() + else: + connection.__init__(":memory:") + assert reference() is None, (method, args, action) + del cursor + connection.close() + + +def callback_self_removal_lifetime(): + for method, args in ( + ("set_trace_callback", ()), + ("set_authorizer", ()), + ("set_progress_handler", (1,)), + ): + connection = sqlite.connect(":memory:") + register = getattr(connection, method) + events = [] + + class Result: + def __bool__(self): + assert events == ["call"] + return False + + class Callback: + def __call__(self, *unused): + register(None, *args) + events.append("call") + return Result() if args else sqlite.SQLITE_OK + + def __del__(self): + events.append(("drop", connection.execute("SELECT 2").fetchone())) + + register(Callback(), *args) + # The callable connection also prepares statements and invokes authorizers. + if method == "set_authorizer": + connection("SELECT 1") + else: + connection.execute("SELECT 1").fetchall() + assert events == ["call", ("drop", (2,))], (method, events) + connection.close() + + +callback_registration_lifetimes() +callback_self_removal_lifetime() From cfdf827fcdc7b57e56a671773a3fc2d19b54e968 Mon Sep 17 00:00:00 2001 From: 1ndahous3 <9205230+1ndahous3@users.noreply.github.com> Date: Thu, 15 Oct 2026 15:27:28 +0300 Subject: [PATCH 2/5] Drain SQLite callback cleanup before reinitialization and after blob operations Assisted-by: Codex:gpt-6 --- crates/stdlib/src/_sqlite3.rs | 265 ++++++++++++++++++++------ extra_tests/snippets/stdlib_sqlite.py | 16 +- 2 files changed, 217 insertions(+), 64 deletions(-) diff --git a/crates/stdlib/src/_sqlite3.rs b/crates/stdlib/src/_sqlite3.rs index c7d1da51cd5..135986b7d5d 100644 --- a/crates/stdlib/src/_sqlite3.rs +++ b/crates/stdlib/src/_sqlite3.rs @@ -15,7 +15,7 @@ mod _sqlite3 { use core::{ ffi::{CStr, c_int, c_longlong, c_uint, c_void}, fmt::Debug, - ops::Deref, + ops::{Deref, DerefMut}, ptr::{NonNull, null, null_mut}, }; use libsqlite3_sys::{ @@ -1075,43 +1075,51 @@ mod _sqlite3 { type Args = ConnectArgs; fn init(zelf: &Py, args: Self::Args, vm: &VirtualMachine) -> PyResult<()> { - with_deferred_drops(|| { - let was_initialized = Radium::swap(&zelf.initialized, false, Ordering::AcqRel); - - // Reset factories to their defaults, matching CPython's behavior. - zelf.reset_factories(vm); + loop { + with_deferred_drops(|| { + Radium::store(&zelf.initialized, false, Ordering::Release); - if was_initialized { - zelf.drop_db(vm)?; + // Reset factories to their defaults, matching CPython's behavior. + zelf.reset_factories(vm); + zelf.drop_db(vm) + })?; + // A finalizer may itself reinitialize the connection. Dispose of + // that database too before publishing the requested replacement. + if zelf.db.lock().is_none() { + break; } + } - // Attempt to open the new database before mutating other state so failures leave - // the connection uninitialized (and subsequent operations raise ProgrammingError). - let db = Self::initialize_db(&args, vm)?; - - let ConnectArgs { - detect_types, - isolation_level, - check_same_thread, - autocommit, - cached_statements, - .. - } = args; - - zelf.statement_cache.lock().capacity = cached_statements.max(0) as usize; - - zelf.detect_types.store(detect_types, Ordering::Relaxed); - zelf.check_same_thread - .store(check_same_thread, Ordering::Relaxed); - *zelf.autocommit.lock() = autocommit; - *zelf.thread_ident.lock() = std::thread::current().id(); - let _ = unsafe { zelf.isolation_level.swap(isolation_level.0) }; - - let mut guard = zelf.db.lock(); - *guard = Some(db); - Radium::store(&zelf.initialized, true, Ordering::Release); - Ok(()) - }) + // Dispose of the old callbacks before publishing the replacement. + // Nested native calls still defer their drops to the outer operation. + // Attempt to open the new database before mutating other state so failures leave + // the connection uninitialized (and subsequent operations raise ProgrammingError). + let db = Self::initialize_db(&args, vm)?; + + let ConnectArgs { + detect_types, + isolation_level, + check_same_thread, + autocommit, + cached_statements, + .. + } = args; + + zelf.statement_cache.lock().capacity = cached_statements.max(0) as usize; + + zelf.detect_types.store(detect_types, Ordering::Relaxed); + zelf.check_same_thread + .store(check_same_thread, Ordering::Relaxed); + *zelf.autocommit.lock() = autocommit; + *zelf.thread_ident.lock() = std::thread::current().id(); + let _ = unsafe { zelf.isolation_level.swap(isolation_level.0) }; + + let previous = zelf.db.lock().replace(db); + Radium::store(&zelf.initialized, true, Ordering::Release); + // An old callback's finalizer can reenter __init__ during teardown. + // Release any database it installed without holding the Rust lock. + with_deferred_drops(|| drop(previous)); + Ok(()) } } @@ -1301,12 +1309,14 @@ mod _sqlite3 { ) }); db.check(ret, vm)?; - drop(db); - let blob = SqliteBlob { blob }; let blob = Blob { connection: zelf, - inner: PyMutex::new(Some(BlobInner { blob, offset: 0 })), + inner: PyMutex::new(Some(BlobInner { + blob, + offset: 0, + db: db.0.clone(), + })), }; Ok(blob.into_ref(&vm.ctx)) } @@ -2837,15 +2847,144 @@ mod _sqlite3 { inner: PyMutex>, } - #[derive(Debug)] struct BlobInner { blob: SqliteBlob, offset: c_int, + db: PyRc, + } + + impl Debug for BlobInner { + fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result { + f.debug_struct("BlobInner") + .field("blob", &self.blob) + .field("offset", &self.offset) + .finish_non_exhaustive() + } + } + + impl BlobInner { + fn check(&self, ret: c_int, vm: &VirtualMachine) -> PyResult<()> { + if ret == SQLITE_OK { + Ok(()) + } else { + Err(self.db.error_extended(vm)) + } + } } impl Drop for BlobInner { fn drop(&mut self) { - with_deferred_drops(|| unsafe { sqlite3_blob_close(self.blob.blob) }); + self.db.pending.lock().blobs.push(self.blob); + self.db.flush_cleanup(); + } + } + + struct BlobGuard<'a> { + inner: Option>, + _db: SqliteOperation, + } + + impl<'a> BlobGuard<'a> { + fn new(inner: PyMappedMutexGuard<'a, BlobInner>) -> Self { + let db = SqliteOperation(inner.db.clone()); + Self { + inner: Some(inner), + _db: db, + } + } + } + + impl Deref for BlobGuard<'_> { + type Target = BlobInner; + + fn deref(&self) -> &Self::Target { + self.inner.as_deref().unwrap() + } + } + + impl DerefMut for BlobGuard<'_> { + fn deref_mut(&mut self) -> &mut Self::Target { + self.inner.as_deref_mut().unwrap() + } + } + + impl Drop for BlobGuard<'_> { + fn drop(&mut self) { + // The operation drains pending cleanup after releasing the blob lock. + self.inner.take(); + } + } + + #[cfg(test)] + #[cfg(feature = "threading")] + mod tests { + use super::*; + use core::time::Duration; + use std::sync::mpsc; + + #[test] + fn blob_operation_drains_contended_cleanup() { + let mut raw = null_mut(); + assert_eq!( + unsafe { + sqlite3_open_v2( + c":memory:".as_ptr(), + &mut raw, + SQLITE_OPEN_READWRITE | SQLITE_OPEN_CREATE | SQLITE_OPEN_FULLMUTEX, + null(), + ) + }, + SQLITE_OK + ); + let db = PyRc::new(Sqlite::from(SqliteRaw { db: raw })); + assert!( + db.exec(c"CREATE TABLE t(b); INSERT INTO t VALUES (zeroblob(1))") + .is_ok() + ); + let open_blob = || { + let mut blob = null_mut(); + assert_eq!( + unsafe { + sqlite3_blob_open( + db.db, + c"main".as_ptr(), + c"t".as_ptr(), + c"b".as_ptr(), + 1, + 0, + &mut blob, + ) + }, + SQLITE_OK + ); + BlobInner { + blob: SqliteBlob { blob }, + offset: 0, + db: db.clone(), + } + }; + let inner = PyMutex::new(Some(open_blob())); + let retired = open_blob(); + let guard = BlobGuard::new(PyMutexGuard::map(inner.lock(), |x| x.as_mut().unwrap())); + let (locked_tx, locked_rx) = mpsc::channel(); + let (release_tx, release_rx) = mpsc::channel(); + let other = db.clone(); + let worker = std::thread::spawn(move || { + let _lock = other.raw.lock(); + locked_tx.send(()).unwrap(); + release_rx.recv_timeout(Duration::from_secs(5)).unwrap(); + }); + locked_rx.recv_timeout(Duration::from_secs(5)).unwrap(); + // Reproduce close() losing both nonblocking cleanup attempts. + db.pending.lock().clear_callbacks = true; + db.flush_cleanup(); + drop(retired); + assert_eq!(db.pending.lock().blobs.len(), 1); + release_tx.send(()).unwrap(); + worker.join().unwrap(); + assert!(db.pending.lock().clear_callbacks); + drop(guard); + assert!(db.pending.lock().is_empty()); } } @@ -2854,7 +2993,8 @@ mod _sqlite3 { #[pymethod] fn close(zelf: &Py) { with_deferred_drops(|| { - zelf.inner.lock().take(); + let inner = zelf.inner.lock().take(); + drop(inner); }); } @@ -2893,7 +3033,7 @@ mod _sqlite3 { let ret = inner .blob .read(buf.as_mut_ptr().cast(), length, inner.offset); - zelf.check(ret, vm)?; + inner.check(ret, vm)?; unsafe { buf.set_len(length as usize) }; inner.offset += length; Ok(vm.ctx.new_bytes(buf)) @@ -2911,7 +3051,7 @@ mod _sqlite3 { inner.blob.write(buf.as_ptr().cast(), length, inner.offset) }); - zelf.check(ret, vm)?; + inner.check(ret, vm)?; inner.offset += length; Ok(()) } @@ -2972,12 +3112,12 @@ mod _sqlite3 { Ok(()) } - fn inner(&self, vm: &VirtualMachine) -> PyResult> { + fn inner(&self, vm: &VirtualMachine) -> PyResult> { let guard = self.inner.lock(); if guard.is_some() { - Ok(PyMutexGuard::map(guard, |x| unsafe { - x.as_mut().unwrap_unchecked() - })) + Ok(BlobGuard::new(PyMutexGuard::map(guard, |x| { + x.as_mut().unwrap() + }))) } else { Err(new_programming_error( vm, @@ -3040,7 +3180,7 @@ mod _sqlite3 { let index = Self::wrapped_index(index?, needle, blob_len, vm)?; let mut byte: u8 = 0; let ret = inner.blob.read_single(&mut byte, index); - self.check(ret, vm).map(|_| vm.ctx.new_int(byte).into()) + inner.check(ret, vm).map(|_| vm.ctx.new_int(byte).into()) } else if let Some(slice) = needle.downcast_ref::() { let blob_len = inner.blob.bytes(); let slice = slice.to_saturated(vm)?; @@ -3053,14 +3193,14 @@ mod _sqlite3 { length as c_int, range.start as c_int, ); - self.check(ret, vm)?; + inner.check(ret, vm)?; unsafe { buf.set_len(length) }; } else { let iter = SaturatedSliceIter::from_adjust_indices(range, step, length); let mut byte: u8 = 0; for index in iter { let ret = inner.blob.read_single(&mut byte, index as c_int); - self.check(ret, vm)?; + inner.check(ret, vm)?; buf.push(byte); } } @@ -3108,7 +3248,7 @@ mod _sqlite3 { return Err(vm.new_value_error("byte must be in range(0, 256)")); } let ret = inner.blob.write_single(val as u8, index); - self.check(ret, vm) + inner.check(ret, vm) } else if let Some(slice) = needle.downcast_ref::() { // Handle slice assignment: blob[a:b:c] = b"..." let Some(value) = value else { @@ -3142,7 +3282,7 @@ mod _sqlite3 { buf.len() as c_int, range.start as c_int, ); - self.check(ret, vm) + inner.check(ret, vm) } else { let span_len = range.end - range.start; let range_start = range.start; @@ -3153,7 +3293,7 @@ mod _sqlite3 { span_len as c_int, range_start as c_int, ); - self.check(ret, vm)?; + inner.check(ret, vm)?; let iter = SaturatedSliceIter::from_adjust_indices(range, step, slice_len); for (i_in_src, abs_idx) in iter.enumerate() { @@ -3165,20 +3305,12 @@ mod _sqlite3 { span_len as c_int, range_start as c_int, ); - self.check(ret, vm) + inner.check(ret, vm) } } else { Err(vm.new_type_error("Blob indices must be integers")) } } - - fn check(&self, ret: c_int, vm: &VirtualMachine) -> PyResult<()> { - if ret == SQLITE_OK { - Ok(()) - } else { - Err(self.connection.db_lock(vm)?.error_extended(vm)) - } - } } impl AsMapping for Blob { @@ -3452,13 +3584,17 @@ mod _sqlite3 { #[derive(Default)] struct SqliteCleanup { statements: Vec, + blobs: Vec, rollback: bool, clear_callbacks: bool, } impl SqliteCleanup { fn is_empty(&self) -> bool { - self.statements.is_empty() && !self.rollback && !self.clear_callbacks + self.statements.is_empty() + && self.blobs.is_empty() + && !self.rollback + && !self.clear_callbacks } } @@ -3509,6 +3645,9 @@ mod _sqlite3 { for statement in pending.statements { unsafe { sqlite3_finalize(statement.st) }; } + for blob in pending.blobs { + unsafe { sqlite3_blob_close(blob.blob) }; + } let callbacks = pending.clear_callbacks.then(|| self.clear_callbacks()); drop(guard); drop(callbacks); diff --git a/extra_tests/snippets/stdlib_sqlite.py b/extra_tests/snippets/stdlib_sqlite.py index 1a3b517ea26..241a9f2e554 100644 --- a/extra_tests/snippets/stdlib_sqlite.py +++ b/extra_tests/snippets/stdlib_sqlite.py @@ -819,13 +819,25 @@ def __init__(self, connection): def __call__(self, *args): return sqlite.SQLITE_OK + def __del__(self): + if action in ("reinit", "reentrant_reinit"): + events.append("drop") + try: + self.connection.close() + except sqlite.ProgrammingError: + pass + if action == "reentrant_reinit" and len(events) == 1: + self.connection.__init__(":memory:") + register(Callback(self.connection), *args) + for method, args in ( ("set_trace_callback", ()), ("set_authorizer", ()), ("set_progress_handler", (1,)), ("set_progress_handler", (0,)), ): - for action in ("disable", "replace", "close", "reinit"): + for action in ("disable", "replace", "close", "reinit", "reentrant_reinit"): + events = [] connection = sqlite.connect(":memory:") cursor = connection.execute("SELECT 1 UNION ALL SELECT 2") register = getattr(connection, method) @@ -842,6 +854,8 @@ def __call__(self, *args): connection.close() else: connection.__init__(":memory:") + assert events == ["drop"] * (2 if action == "reentrant_reinit" else 1) + assert connection.execute("SELECT 1").fetchone() == (1,) assert reference() is None, (method, args, action) del cursor connection.close() From 36418c4a83aba23683d995bbaae5053f4f17b17b Mon Sep 17 00:00:00 2001 From: 1ndahous3 <9205230+1ndahous3@users.noreply.github.com> Date: Thu, 15 Oct 2026 17:25:56 +0300 Subject: [PATCH 3/5] Drain SQLite cleanup after statement reset Assisted-by: Codex:gpt-6 --- crates/stdlib/src/_sqlite3.rs | 6 ++- extra_tests/snippets/stdlib_sqlite.py | 67 +++++++++++++++++++++++++++ 2 files changed, 72 insertions(+), 1 deletion(-) diff --git a/crates/stdlib/src/_sqlite3.rs b/crates/stdlib/src/_sqlite3.rs index 135986b7d5d..89e145dc96b 100644 --- a/crates/stdlib/src/_sqlite3.rs +++ b/crates/stdlib/src/_sqlite3.rs @@ -3513,11 +3513,14 @@ mod _sqlite3 { fn drop(&mut self) { with_deferred_drops(|| { let statement = self.0.lock(); + let db = SqliteOperation(statement.db.clone()); statement.reset(); // Cached statements must not retain previous parameter buffers. unsafe { sqlite3_clear_bindings(statement.st) }; drop(statement); self.0.in_use.store(false, Ordering::Release); + // Reset can delay cleanup queued by another thread's BLOB close. + drop(db); }); } } @@ -3662,13 +3665,14 @@ mod _sqlite3 { fn close_callbacks(&self, rollback: bool, vm: &VirtualMachine) -> PyResult<()> { self.callbacks.lock().closed = true; - if let Some(_guard) = self.raw.try_lock() { + if let Some(guard) = self.raw.try_lock() { let result = if rollback && !self.is_autocommit() { self._exec(c"ROLLBACK", vm) } else { Ok(()) }; self.pending.lock().clear_callbacks = true; + drop(guard); self.flush_cleanup(); result } else { diff --git a/extra_tests/snippets/stdlib_sqlite.py b/extra_tests/snippets/stdlib_sqlite.py index 241a9f2e554..8c3e00da6f8 100644 --- a/extra_tests/snippets/stdlib_sqlite.py +++ b/extra_tests/snippets/stdlib_sqlite.py @@ -837,6 +837,8 @@ def __del__(self): ("set_progress_handler", (0,)), ): for action in ("disable", "replace", "close", "reinit", "reentrant_reinit"): + if args == (0,) and action != "disable": + continue events = [] connection = sqlite.connect(":memory:") cursor = connection.execute("SELECT 1 UNION ALL SELECT 2") @@ -897,3 +899,68 @@ def __del__(self): callback_registration_lifetimes() callback_self_removal_lifetime() + + +def blob_close_during_cursor_reset(): + import tempfile + import threading + + with tempfile.TemporaryDirectory() as directory: + path = directory + "/blob.db" + connection = sqlite.connect(path, isolation_level=None, check_same_thread=False) + connection.execute("CREATE TABLE t(value)") + connection.execute("INSERT INTO t VALUES (zeroblob(1)), (zeroblob(1))") + resetting = threading.Event() + release = threading.Event() + cursor_closed = threading.Event() + blob_closed = threading.Event() + failures = [] + + class Window: + def step(self, value): + pass + + def value(self): + return 0 + + def finalize(self): + resetting.set() + if not release.wait(5): + failures.append("cursor reset timed out") + return 0 + + connection.create_window_function("hold", 1, Window) + blob = connection.blobopen("t", "value", 1) + cursor = connection.execute( + "SELECT hold(rowid) OVER (ROWS UNBOUNDED PRECEDING) FROM t" + ) + + def close(resource, finished): + resource.close() + finished.set() + + resetter = threading.Thread( + target=close, args=(cursor, cursor_closed), daemon=True + ) + closer = threading.Thread(target=close, args=(blob, blob_closed), daemon=True) + resetter.start() + assert resetting.wait(5) + closer.start() + # CPython waits for reset; RustPython may queue the BLOB for later cleanup. + blob_closed.wait(0.2) + release.set() + resetter.join(5) + closer.join(5) + assert not resetter.is_alive() and not closer.is_alive() + assert cursor_closed.is_set() and blob_closed.is_set() + assert not failures, failures + other = sqlite.connect(path, timeout=0, isolation_level=None) + try: + # No intervening call on connection may be needed to release its lock. + other.execute("UPDATE t SET value=zeroblob(1)") + finally: + other.close() + connection.close() + + +blob_close_during_cursor_reset() From 1bee2b2a751a5fe570a4d4583c010d27f41782ac Mon Sep 17 00:00:00 2001 From: 1ndahous3 <9205230+1ndahous3@users.noreply.github.com> Date: Fri, 16 Oct 2026 01:16:12 +0300 Subject: [PATCH 4/5] Fix concurrent SQLite connection reinitialization Assisted-by: Codex:gpt-6 --- crates/stdlib/src/_sqlite3.rs | 54 +++++++++++++++++++-------- extra_tests/snippets/stdlib_sqlite.py | 51 +++++++++++++++++++++++++ 2 files changed, 90 insertions(+), 15 deletions(-) diff --git a/crates/stdlib/src/_sqlite3.rs b/crates/stdlib/src/_sqlite3.rs index 89e145dc96b..e432a6a8102 100644 --- a/crates/stdlib/src/_sqlite3.rs +++ b/crates/stdlib/src/_sqlite3.rs @@ -1105,21 +1105,45 @@ mod _sqlite3 { .. } = args; - zelf.statement_cache.lock().capacity = cached_statements.max(0) as usize; - - zelf.detect_types.store(detect_types, Ordering::Relaxed); - zelf.check_same_thread - .store(check_same_thread, Ordering::Relaxed); - *zelf.autocommit.lock() = autocommit; - *zelf.thread_ident.lock() = std::thread::current().id(); - let _ = unsafe { zelf.isolation_level.swap(isolation_level.0) }; - - let previous = zelf.db.lock().replace(db); - Radium::store(&zelf.initialized, true, Ordering::Release); - // An old callback's finalizer can reenter __init__ during teardown. - // Release any database it installed without holding the Rust lock. - with_deferred_drops(|| drop(previous)); - Ok(()) + loop { + let mut current = zelf.db.lock(); + if current.is_none() { + let previous = unsafe { zelf.isolation_level.swap(isolation_level.0.clone()) }; + if previous.as_ref().is_some_and(|previous| { + !isolation_level + .0 + .as_ref() + .is_some_and(|requested| previous.is(requested)) + }) { + // A str subclass can finalize too. Keep the requested + // value alive, but retire other values before publication. + drop(current); + with_deferred_drops(|| drop(previous)); + continue; + } + + // No destructor-capable owner remains to release here. + // Publish the settings and database under the same lock. + zelf.statement_cache.lock().capacity = cached_statements.max(0) as usize; + zelf.detect_types.store(detect_types, Ordering::Relaxed); + zelf.check_same_thread + .store(check_same_thread, Ordering::Relaxed); + *zelf.autocommit.lock() = autocommit; + *zelf.thread_ident.lock() = std::thread::current().id(); + *current = Some(db); + Radium::store(&zelf.initialized, true, Ordering::Release); + return Ok(()); + } + drop(current); + + // A concurrent initializer or finalizer installed another database. + // Finish its teardown outside the lock before trying to publish. + with_deferred_drops(|| { + Radium::store(&zelf.initialized, false, Ordering::Release); + zelf.reset_factories(vm); + zelf.drop_db(vm) + })?; + } } } diff --git a/extra_tests/snippets/stdlib_sqlite.py b/extra_tests/snippets/stdlib_sqlite.py index 8c3e00da6f8..a6e9e66d971 100644 --- a/extra_tests/snippets/stdlib_sqlite.py +++ b/extra_tests/snippets/stdlib_sqlite.py @@ -964,3 +964,54 @@ def close(resource, finished): blob_close_during_cursor_reset() + + +def reinit_during_isolation_cleanup(): + import threading + + failures = [] + armed = False + + class Callback: + def __call__(self, sql): + pass + + def __del__(self): + try: + connection.close() + except Exception as exc: + failures.append(exc) + + def replace(): + try: + connection.__init__(":memory:", check_same_thread=False, autocommit=True) + connection.set_trace_callback(Callback()) + except Exception as exc: + failures.append(exc) + + class Level(str): + def __del__(self): + # CPython releases this subclass during argument conversion; + # RustPython retains it until the isolation level is replaced. + if armed: + worker = threading.Thread(target=replace, daemon=True) + worker.start() + worker.join(5) + if worker.is_alive(): + failures.append("reentrant initializer blocked") + + connection = sqlite.connect( + ":memory:", isolation_level=Level(""), check_same_thread=False + ) + armed = True + try: + connection.__init__(":memory:", check_same_thread=False, autocommit=False) + assert not failures, failures + assert connection.execute("SELECT 1").fetchone() == (1,) + assert connection.autocommit is False + assert connection.in_transaction + finally: + connection.close() + + +reinit_during_isolation_cleanup() From a9c7a56bf92854e62441a36a944295a4a215e8de Mon Sep 17 00:00:00 2001 From: 1ndahous3 <9205230+1ndahous3@users.noreply.github.com> Date: Fri, 16 Oct 2026 01:54:17 +0300 Subject: [PATCH 5/5] Restore SQLite factory defaults after callback cleanup Assisted-by: Codex:gpt-6 --- crates/stdlib/src/_sqlite3.rs | 14 ++++++++++---- extra_tests/snippets/stdlib_sqlite.py | 27 +++++++++++++++++++++++++++ 2 files changed, 37 insertions(+), 4 deletions(-) diff --git a/crates/stdlib/src/_sqlite3.rs b/crates/stdlib/src/_sqlite3.rs index e432a6a8102..0fc0454af13 100644 --- a/crates/stdlib/src/_sqlite3.rs +++ b/crates/stdlib/src/_sqlite3.rs @@ -1105,20 +1105,26 @@ mod _sqlite3 { .. } = args; + let default_text_factory = PyStr::class(&vm.ctx).to_owned().into_object(); loop { let mut current = zelf.db.lock(); if current.is_none() { let previous = unsafe { zelf.isolation_level.swap(isolation_level.0.clone()) }; + let row_factory = unsafe { zelf.row_factory.swap(None) }; + let text_factory = + unsafe { zelf.text_factory.swap(default_text_factory.clone()) }; if previous.as_ref().is_some_and(|previous| { !isolation_level .0 .as_ref() .is_some_and(|requested| previous.is(requested)) - }) { - // A str subclass can finalize too. Keep the requested - // value alive, but retire other values before publication. + }) || row_factory.is_some() + || !text_factory.is(&default_text_factory) + { + // Factories and str subclasses can finalize too. Retire + // them before publication, then recheck any reentrant changes. drop(current); - with_deferred_drops(|| drop(previous)); + with_deferred_drops(|| drop((previous, row_factory, text_factory))); continue; } diff --git a/extra_tests/snippets/stdlib_sqlite.py b/extra_tests/snippets/stdlib_sqlite.py index a6e9e66d971..684a3765e2c 100644 --- a/extra_tests/snippets/stdlib_sqlite.py +++ b/extra_tests/snippets/stdlib_sqlite.py @@ -1015,3 +1015,30 @@ def __del__(self): reinit_during_isolation_cleanup() + + +def reinit_restores_factories_after_callback_cleanup(): + connection = sqlite.connect(":memory:") + events = [] + + class Callback: + def __call__(self, sql): + pass + + def __del__(self): + connection.row_factory = sqlite.Row + connection.text_factory = bytes + events.append("factories changed") + + connection.set_trace_callback(Callback()) + try: + connection.__init__(":memory:") + assert events == ["factories changed"] + assert connection.row_factory is None + assert connection.text_factory is str + assert connection.execute("SELECT 'hello'").fetchone() == ("hello",) + finally: + connection.close() + + +reinit_restores_factories_after_callback_cleanup()