Release SQLite callbacks and expanded trace SQL safely (#9151)

* Release SQLite callbacks and expanded trace SQL safely

Assisted-by: Codex:gpt-6

* Drain SQLite callback cleanup before reinitialization and after blob operations

Assisted-by: Codex:gpt-6

* Drain SQLite cleanup after statement reset

Assisted-by: Codex:gpt-6

* Fix concurrent SQLite connection reinitialization

Assisted-by: Codex:gpt-6

* Restore SQLite factory defaults after callback cleanup

Assisted-by: Codex:gpt-6

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::{
@@ -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<Self> {
+            let data = data.cast::<Self>();
+            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::<Self>() };
+            let data = unsafe { Self::registered(data) };
             data.with_vm(|callable, vm| {
                 let f = || -> PyResult<c_int> {
                     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::<Self>() };
+            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::<Self>() };
+            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<Self>, args: Self::Args, vm: &VirtualMachine) -> PyResult {
+            with_deferred_drops(|| Self::call_inner(zelf, args, vm))
+        }
+    }
+
+    impl Connection {
+        fn call_inner(zelf: &Py<Self>, args: FuncArgs, vm: &VirtualMachine) -> PyResult {
             let _ = zelf.db_lock(vm)?;
 
             let (sql,): (PyUtf8StrRef,) = args.bind(vm)?;
@@ -1052,43 +1075,81 @@ mod _sqlite3 {
         type Args = ConnectArgs;
 
         fn init(zelf: &Py<Self>, args: Self::Args, vm: &VirtualMachine) -> PyResult<()> {
-            with_deferred_drops(|| {
-                let was_initialized = Radium::swap(&zelf.initialized, false, Ordering::AcqRel);
+            loop {
+                with_deferred_drops(|| {
+                    Radium::store(&zelf.initialized, false, Ordering::Release);
 
-                // Reset factories to their defaults, matching CPython's behavior.
-                zelf.reset_factories(vm);
-
-                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)?;
+            // 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;
+            let ConnectArgs {
+                detect_types,
+                isolation_level,
+                check_same_thread,
+                autocommit,
+                cached_statements,
+                ..
+            } = args;
 
-                zelf.statement_cache.lock().capacity = cached_statements.max(0) as usize;
+            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))
+                    }) || 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, row_factory, text_factory)));
+                        continue;
+                    }
 
-                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) };
+                    // 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);
 
-                let mut guard = zelf.db.lock();
-                *guard = Some(db);
-                Radium::store(&zelf.initialized, true, Ordering::Release);
-                Ok(())
-            })
+                // 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)
+                })?;
+            }
         }
     }
 
@@ -1158,10 +1219,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(())
         }
@@ -1280,12 +1339,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))
         }
@@ -1683,22 +1744,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 +1753,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 +1763,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(());
-            };
+            zelf.set_callback(callable, CallbackKind::Progress(n), vm)
+        }
 
-            unsafe {
-                sqlite3_progress_handler(
-                    db.db,
-                    n,
-                    Some(CallbackData::progress_callback),
-                    Box::into_raw(Box::new(data)).cast(),
-                )
-            };
-
-            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]
@@ -2815,15 +2877,144 @@ mod _sqlite3 {
         inner: PyMutex<Option<BlobInner>>,
     }
 
-    #[derive(Debug)]
     struct BlobInner {
         blob: SqliteBlob,
         offset: c_int,
+        db: PyRc<Sqlite>,
+    }
+
+    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<PyMappedMutexGuard<'a, BlobInner>>,
+        _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());
         }
     }
 
@@ -2832,7 +3023,8 @@ mod _sqlite3 {
         #[pymethod]
         fn close(zelf: &Py<Self>) {
             with_deferred_drops(|| {
-                zelf.inner.lock().take();
+                let inner = zelf.inner.lock().take();
+                drop(inner);
             });
         }
 
@@ -2871,7 +3063,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))
@@ -2889,7 +3081,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(())
         }
@@ -2950,12 +3142,12 @@ mod _sqlite3 {
             Ok(())
         }
 
-        fn inner(&self, vm: &VirtualMachine) -> PyResult<PyMappedMutexGuard<'_, BlobInner>> {
+        fn inner(&self, vm: &VirtualMachine) -> PyResult<BlobGuard<'_>> {
             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,
@@ -3018,7 +3210,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::<PySlice>() {
                 let blob_len = inner.blob.bytes();
                 let slice = slice.to_saturated(vm)?;
@@ -3031,14 +3223,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);
                     }
                 }
@@ -3086,7 +3278,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::<PySlice>() {
                 // Handle slice assignment: blob[a:b:c] = b"..."
                 let Some(value) = value else {
@@ -3120,7 +3312,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;
@@ -3131,7 +3323,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() {
@@ -3143,20 +3335,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 {
@@ -3359,11 +3543,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);
             });
         }
     }
@@ -3371,17 +3558,76 @@ mod _sqlite3 {
     struct Sqlite {
         raw: SqliteRaw,
         pending: PyMutex<SqliteCleanup>,
+        callbacks: PyMutex<CallbackSlots>,
+    }
+
+    #[derive(Default)]
+    struct CallbackSlots {
+        trace: Option<PyRc<CallbackData>>,
+        progress: Option<PyRc<CallbackData>>,
+        authorizer: Option<PyRc<CallbackData>>,
+        closed: bool,
+    }
+
+    #[derive(Clone, Copy)]
+    enum CallbackKind {
+        Trace,
+        Progress(c_int),
+        Authorizer,
+    }
+
+    impl CallbackKind {
+        fn slot(self, slots: &mut CallbackSlots) -> &mut Option<PyRc<CallbackData>> {
+            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<SqliteStatementRaw>,
+        blobs: Vec<SqliteBlob>,
         rollback: bool,
+        clear_callbacks: bool,
     }
 
     impl SqliteCleanup {
         fn is_empty(&self) -> bool {
-            self.statements.is_empty() && !self.rollback
+            self.statements.is_empty()
+                && self.blobs.is_empty()
+                && !self.rollback
+                && !self.clear_callbacks
         }
     }
 
@@ -3390,6 +3636,7 @@ mod _sqlite3 {
             Self {
                 raw,
                 pending: PyMutex::new(SqliteCleanup::default()),
+                callbacks: PyMutex::new(CallbackSlots::default()),
             }
         }
     }
@@ -3431,7 +3678,12 @@ 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);
                     // 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 +3693,45 @@ mod _sqlite3 {
             });
         }
 
-        fn rollback_on_close(&self, vm: &VirtualMachine) -> PyResult<()> {
-            if let Some(_guard) = self.raw.try_lock() {
-                if !self.is_autocommit() {
-                    self._exec(c"ROLLBACK", vm)?;
-                }
+        fn close_callbacks(&self, rollback: bool, vm: &VirtualMachine) -> PyResult<()> {
+            self.callbacks.lock().closed = true;
+            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 {
-                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<PyRc<CallbackData>>; 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 +3803,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) };
+            });
         }
     }
 

extra_tests/snippets/stdlib_sqlite.py

@@ -809,3 +809,236 @@ def concurrent_sqlite_transactions():
 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
+
+        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", "reentrant_reinit"):
+            if args == (0,) and action != "disable":
+                continue
+            events = []
+            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 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()
+
+
+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()
+
+
+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()
+
+
+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()
+
+
+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()