diff --git a/doc/api/sqlite.md b/doc/api/sqlite.md
index e50a66e92621..da7b4e12d8cd 100644
--- a/doc/api/sqlite.md
+++ b/doc/api/sqlite.md
@@ -442,6 +442,11 @@ wrapper around [`sqlite3_create_function_v2()`][].
 
 <!-- YAML
 added: v24.10.0
+changes:
+  - version: REPLACEME
+    pr-url: https://github.com/nodejs/node/pull/65156
+    description: Accessing the invoking database connection from the authorizer
+                 callback now throws.
 -->
 
 * `callback` {Function|null} The authorizer function to set, or `null` to
@@ -467,6 +472,31 @@ The callback must return one of the following constants:
 * `SQLITE_DENY` - Deny the operation (causes an error).
 * `SQLITE_IGNORE` - Ignore the operation (silently skip).
 
+SQLite requires that the authorizer callback not modify the database connection
+that invoked it, which includes preparing and stepping statements. Methods that
+would do so throw an error with code `ERR_INVALID_STATE` while the callback is
+on the stack, including `database.prepare()`, `database.exec()`, the execution
+methods of that connection's statements, iterators, and tag stores, and
+`database.setAuthorizer()` itself. Other connections remain usable.
+
+The callback can also be invoked from within `statement.run()`,
+`statement.get()`, and similar methods, because SQLite may re-prepare a
+statement during execution after a schema change.
+
+Separately, a statement that is currently being executed cannot be reentered.
+Calling `statement.close()` on it would free the virtual machine that is
+running, and re-running it through `statement.run()`, `statement.get()`,
+`statement.all()`, `statement.iterate()`, `iterator.next()`,
+`iterator.return()`, or the equivalent tag store methods would reset that
+virtual machine mid-execution. All of these throw an `ERR_INVALID_STATE` error
+instead. This applies to any callback SQLite invokes during execution, such as a
+user-defined function. Other statements on the connection remain usable.
+
+Operations that touch no SQLite state stay available from the callback:
+`sqlTagStore.clear()`, which only drops cached statements, and `next()` and
+`return()` on an already-drained iterator, which keep returning
+`{ done: true }`.
+
 ```cjs
 const { DatabaseSync, constants } = require('node:sqlite');
 const db = new DatabaseSync(':memory:');
@@ -674,6 +704,9 @@ console.log(query.get());
 <!-- YAML
 added: v22.5.0
 changes:
+  - version: REPLACEME
+    pr-url: https://github.com/nodejs/node/pull/62757
+    description: Add the `persistent` option.
   - version: REPLACEME
     pr-url: https://github.com/nodejs/node/pull/65157
     description: Throw `ERR_INVALID_ARG_VALUE` if `sql` contains no statements.
@@ -690,10 +723,14 @@ changes:
     database options or `true`.
   * `allowUnknownNamedParameters` {boolean} If `true`, unknown named parameters
     are ignored. **Default:** inherited from database options or `false`.
+  * `persistent` {boolean} If `true`, hints to SQLite that this statement will
+    be retained for a long time and likely reused many times. SQLite currently
+    responds to this hint by avoiding lookaside memory. Corresponds to the
+    [`SQLITE_PREPARE_PERSISTENT`][] flag. **Default:** `false`.
 * Returns: {StatementSync} The prepared statement.
 
 Compiles a SQL statement into a [prepared statement][]. This method is a wrapper
-around [`sqlite3_prepare_v2()`][].
+around [`sqlite3_prepare_v3()`][].
 
 ### `database.createTagStore([maxSize])`
 
@@ -1860,6 +1897,7 @@ callback function to indicate what type of operation is being authorized.
 [`SQLITE_DETERMINISTIC`]: https://www.sqlite.org/c3ref/c_deterministic.html
 [`SQLITE_DIRECTONLY`]: https://www.sqlite.org/c3ref/c_deterministic.html
 [`SQLITE_MAX_FUNCTION_ARG`]: https://www.sqlite.org/limits.html#max_function_arg
+[`SQLITE_PREPARE_PERSISTENT`]: https://sqlite.org/c3ref/c_prepare_dont_log.html#sqlitepreparepersistent
 [`SQLTagStore`]: #class-sqltagstore
 [`database.applyChangeset()`]: #databaseapplychangesetchangeset-options
 [`database.createTagStore()`]: #databasecreatetagstoremaxsize
@@ -1885,7 +1923,7 @@ callback function to indicate what type of operation is being authorized.
 [`sqlite3_get_autocommit()`]: https://sqlite.org/c3ref/get_autocommit.html
 [`sqlite3_last_insert_rowid()`]: https://www.sqlite.org/c3ref/last_insert_rowid.html
 [`sqlite3_load_extension()`]: https://www.sqlite.org/c3ref/load_extension.html
-[`sqlite3_prepare_v2()`]: https://www.sqlite.org/c3ref/prepare.html
+[`sqlite3_prepare_v3()`]: https://www.sqlite.org/c3ref/prepare.html
 [`sqlite3_serialize()`]: https://sqlite.org/c3ref/serialize.html
 [`sqlite3_set_authorizer()`]: https://sqlite.org/c3ref/set_authorizer.html
 [`sqlite3_sql()`]: https://www.sqlite.org/c3ref/expanded_sql.html
diff --git a/lib/internal/streams/readable.js b/lib/internal/streams/readable.js
index 25819d14b7e8..2b87b7952bf9 100644
--- a/lib/internal/streams/readable.js
+++ b/lib/internal/streams/readable.js
@@ -23,6 +23,8 @@
 
 const {
   ArrayPrototypeIndexOf,
+  AsyncIteratorPrototype,
+  FunctionPrototypeCall,
   NumberIsInteger,
   NumberIsNaN,
   NumberParseInt,
@@ -30,6 +32,9 @@ const {
   ObjectKeys,
   ObjectSetPrototypeOf,
   Promise,
+  PromisePrototypeThen,
+  PromiseReject,
+  PromiseResolve,
   ReflectApply,
   SafeSet,
   Symbol,
@@ -100,6 +105,7 @@ const FastBuffer = Buffer[SymbolSpecies];
 
 const { StringDecoder } = require('string_decoder');
 const from = require('internal/streams/from');
+const FixedQueue = require('internal/fixed_queue');
 
 ObjectSetPrototypeOf(Readable.prototype, Stream.prototype);
 ObjectSetPrototypeOf(Readable, Stream);
@@ -1386,10 +1392,22 @@ function streamToAsyncIterator(stream, options) {
   return iter;
 }
 
-async function* createAsyncIterator(stream, options) {
+// Async iterator over a Readable. Requests received while another is
+// outstanding are queued and processed in order.
+function createAsyncIterator(stream, options) {
   let callback = nop;
-
-  function next(resolve) {
+  let error;            // undefined: active, null: ended cleanly, else: Error
+  let started = false;
+  let completed = false;
+  let inFlight = false; // An asynchronous request is outstanding
+  let queue = null;     // Requests received while inFlight
+  let draining = false;
+  let cleanup;
+
+  // Used both as the 'readable' listener (where `this === stream`) and
+  // as a promise executor storing the resolver that wakes up a pending
+  // pump().
+  function wakeup(resolve) {
     if (this === stream) {
       callback();
       callback = nop;
@@ -1398,32 +1416,23 @@ async function* createAsyncIterator(stream, options) {
     }
   }
 
-  stream.on('readable', next);
+  function start() {
+    started = true;
 
-  let error;
-  const cleanup = eos(stream, { writable: false }, (err) => {
-    error = err ? aggregateTwoErrors(error, err) : null;
-    callback();
-    callback = nop;
-  });
+    stream.on('readable', wakeup);
+
+    cleanup = eos(stream, { writable: false }, (err) => {
+      error = err ? aggregateTwoErrors(error, err) : null;
+      callback();
+      callback = nop;
+    });
+  }
+
+  // Complete the iterator and either destroy the stream or detach
+  // from it.
+  function finalize() {
+    completed = true;
 
-  try {
-    while (true) {
-      const chunk = stream.destroyed ? null : stream.read();
-      if (chunk !== null) {
-        yield chunk;
-      } else if (error) {
-        throw error;
-      } else if (error === null) {
-        return;
-      } else {
-        await new Promise(next);
-      }
-    }
-  } catch (err) {
-    error = aggregateTwoErrors(error, err);
-    throw error;
-  } finally {
     const preserveHalfOpenDuplex =
       error === null &&
       stream.allowHalfOpen === true &&
@@ -1437,10 +1446,188 @@ async function* createAsyncIterator(stream, options) {
     ) {
       destroyImpl.destroyer(stream, null);
     } else {
-      stream.off('readable', next);
+      stream.off('readable', wakeup);
       cleanup();
     }
   }
+
+  function settleError(err, reject) {
+    error = aggregateTwoErrors(error, err);
+    finalize();
+    reject(error);
+  }
+
+  function drain() {
+    // Requests settled synchronously call back into drain(); the guard
+    // keeps a single loop going instead of recursing once per request.
+    if (draining) {
+      return;
+    }
+    draining = true;
+    try {
+      while (!inFlight && !queue.isEmpty()) {
+        const req = queue.shift();
+        if (req.type === 'next') {
+          processNext(req.resolve, req.reject);
+        } else if (req.type === 'return') {
+          processReturn(req.value, req.resolve);
+        } else {
+          processThrow(req.value, req.reject);
+        }
+      }
+    } finally {
+      draining = false;
+    }
+  }
+
+  // Thenable chunks are unwrapped before delivery; a rejection tears
+  // down the iterator and the stream.
+  function onChunkFulfilled(value) {
+    inFlight = false;
+    if (queue !== null) drain();
+    return { done: false, value };
+  }
+
+  function onChunkRejected(err) {
+    inFlight = false;
+    error = aggregateTwoErrors(error, err);
+    finalize();
+    if (queue !== null) drain();
+    throw error;
+  }
+
+  // Runs with inFlight === true; settles the request and hands over to
+  // any requests that queued up behind it.
+  function pump(resolve, reject) {
+    const chunk = stream.destroyed ? null : stream.read();
+    if (chunk !== null) {
+      // Read `then` only once so that a getter cannot observe (or throw
+      // on) a second access.
+      const then = chunk.then;
+      if (typeof then === 'function') {
+        FunctionPrototypeCall(then, chunk, (value) => {
+          inFlight = false;
+          resolve({ done: false, value });
+          if (queue !== null) drain();
+        }, (err) => {
+          inFlight = false;
+          settleError(err, reject);
+          if (queue !== null) drain();
+        });
+        return;
+      }
+      inFlight = false;
+      resolve({ done: false, value: chunk });
+      if (queue !== null) drain();
+    } else if (error) {
+      inFlight = false;
+      settleError(error, reject);
+      if (queue !== null) drain();
+    } else if (error === null) {
+      inFlight = false;
+      finalize();
+      resolve({ done: true, value: undefined });
+      if (queue !== null) drain();
+    } else {
+      // No data buffered yet; wait for 'readable' or end-of-stream and
+      // retry.
+      PromisePrototypeThen(new Promise(wakeup), () => pump(resolve, reject));
+    }
+  }
+
+  function processNext(resolve, reject) {
+    if (completed) {
+      resolve({ done: true, value: undefined });
+      return;
+    }
+    if (!started) start();
+    inFlight = true;
+    pump(resolve, reject);
+  }
+
+  function processReturn(value, resolve) {
+    if (!completed) {
+      if (started) {
+        finalize();
+      } else {
+        // Never started: complete without touching the stream.
+        completed = true;
+      }
+    }
+    resolve({ done: true, value });
+  }
+
+  function processThrow(err, reject) {
+    if (completed || !started) {
+      completed = true;
+      reject(err);
+      return;
+    }
+    settleError(err, reject);
+  }
+
+  return {
+    __proto__: AsyncIteratorPrototype,
+    next() {
+      if (!inFlight && !completed) {
+        if (!started) start();
+        // Fast path: a chunk is already buffered.
+        const chunk = stream.destroyed ? null : stream.read();
+        if (chunk !== null) {
+          // Read `then` only once so that a getter cannot observe (or
+          // throw on) a second access.
+          const then = chunk.then;
+          if (typeof then === 'function') {
+            inFlight = true;
+            return FunctionPrototypeCall(
+              then, chunk, onChunkFulfilled, onChunkRejected);
+          }
+          return PromiseResolve({ done: false, value: chunk });
+        }
+        if (error) {
+          finalize();
+          return PromiseReject(error);
+        }
+        if (error === null) {
+          finalize();
+          return PromiseResolve({ done: true, value: undefined });
+        }
+        // No data buffered yet; wait for 'readable' or end-of-stream.
+        inFlight = true;
+        return new Promise((resolve, reject) => {
+          PromisePrototypeThen(new Promise(wakeup), () => pump(resolve, reject));
+        });
+      }
+      return new Promise((resolve, reject) => {
+        if (inFlight) {
+          queue ??= new FixedQueue();
+          queue.push({ __proto__: null, type: 'next', value: undefined, resolve, reject });
+        } else {
+          resolve({ done: true, value: undefined });
+        }
+      });
+    },
+    return(value) {
+      return new Promise((resolve, reject) => {
+        if (inFlight) {
+          queue ??= new FixedQueue();
+          queue.push({ __proto__: null, type: 'return', value, resolve, reject });
+        } else {
+          processReturn(value, resolve);
+        }
+      });
+    },
+    throw(err) {
+      return new Promise((resolve, reject) => {
+        if (inFlight) {
+          queue ??= new FixedQueue();
+          queue.push({ __proto__: null, type: 'throw', value: err, resolve, reject });
+        } else {
+          processThrow(err, reject);
+        }
+      });
+    },
+  };
 }
 
 let composeImpl;
diff --git a/src/node_sqlite.cc b/src/node_sqlite.cc
index 156d4f2e5f5c..1b7c11cb69e7 100644
--- a/src/node_sqlite.cc
+++ b/src/node_sqlite.cc
@@ -107,6 +107,27 @@ inline MaybeLocal<String> Utf8StringMaybeOneByte(Isolate* isolate,
     }                                                                          \
   } while (0)
 
+// SQLite requires that an authorizer callback not modify the connection that
+// invoked it. Preparing and stepping statements both count as modifying it.
+// See https://www.sqlite.org/c3ref/set_authorizer.html.
+#define THROW_AND_RETURN_IF_IN_AUTHORIZER(env, db)                             \
+  THROW_AND_RETURN_ON_BAD_STATE(                                               \
+      (env),                                                                   \
+      (db)->IsInAuthorizerCallback(),                                          \
+      "database cannot be accessed from an authorizer callback")
+
+// A statement's virtual machine cannot be reentered while sqlite3_step() is
+// running it. Finalizing it frees the VM outright, and re-running it resets the
+// VM mid-execution; both are use-after-free rather than merely a contract
+// violation. Callbacks that SQLite invokes during execution are therefore
+// barred from reaching the statement being stepped, though other statements on
+// the connection stay usable.
+#define THROW_AND_RETURN_IF_STEPPING(env, stmt)                                \
+  THROW_AND_RETURN_ON_BAD_STATE(                                               \
+      (env),                                                                   \
+      (stmt)->db_->IsSteppingStatement((stmt)->statement_.get()),              \
+      "statement is already being executed")
+
 #define SQLITE_VALUE_TO_JS(from, isolate, use_big_int_args, result, ...)       \
   do {                                                                         \
     switch (sqlite3_##from##_type(__VA_ARGS__)) {                              \
@@ -840,6 +861,12 @@ Intercepted DatabaseSyncLimits::LimitsSetter(
     return Intercepted::kYes;
   }
 
+  if (limits->database_->IsInAuthorizerCallback()) {
+    THROW_ERR_INVALID_STATE(
+        env, "database cannot be accessed from an authorizer callback");
+    return Intercepted::kYes;
+  }
+
   if (!value->IsNumber()) {
     THROW_ERR_INVALID_ARG_TYPE(
         isolate, "Limit value must be a non-negative integer or Infinity.");
@@ -1097,6 +1124,7 @@ void DatabaseSync::CreateTagStore(const FunctionCallbackInfo<Value>& args) {
     THROW_ERR_INVALID_STATE(env, "database is not open");
     return;
   }
+  THROW_AND_RETURN_IF_IN_AUTHORIZER(env, db);
   int capacity = 1000;
   if (args.Length() > 0 && !args[0]->IsUndefined()) {
     if (!args[0]->IsNumber()) {
@@ -1499,6 +1527,7 @@ void DatabaseSync::Prepare(const FunctionCallbackInfo<Value>& args) {
   ASSIGN_OR_RETURN_UNWRAP(&db, args.This());
   Environment* env = Environment::GetCurrent(args);
   THROW_AND_RETURN_ON_BAD_STATE(env, !db->IsOpen(), "database is not open");
+  THROW_AND_RETURN_IF_IN_AUTHORIZER(env, db);
 
   if (!args[0]->IsString()) {
     THROW_ERR_INVALID_ARG_TYPE(env->isolate(),
@@ -1510,6 +1539,7 @@ void DatabaseSync::Prepare(const FunctionCallbackInfo<Value>& args) {
   std::optional<bool> use_big_ints;
   std::optional<bool> allow_bare_named_params;
   std::optional<bool> allow_unknown_named_params;
+  std::optional<bool> persistent;
 
   if (args.Length() > 1 && !args[1]->IsUndefined()) {
     if (!args[1]->IsObject()) {
@@ -1590,11 +1620,34 @@ void DatabaseSync::Prepare(const FunctionCallbackInfo<Value>& args) {
       }
       allow_unknown_named_params = allow_unknown_named_params_v->IsTrue();
     }
+
+    Local<Value> persistent_v;
+    if (!options
+             ->Get(env->context(),
+                   FIXED_ONE_BYTE_STRING(env->isolate(), "persistent"))
+             .ToLocal(&persistent_v)) {
+      return;
+    }
+    if (!persistent_v->IsUndefined()) {
+      if (!persistent_v->IsBoolean()) {
+        THROW_ERR_INVALID_ARG_TYPE(
+            env->isolate(),
+            "The \"options.persistent\" argument must be a boolean.");
+        return;
+      }
+      persistent = persistent_v->IsTrue();
+    }
   }
 
   Utf8Value sql(env->isolate(), args[0].As<String>());
   sqlite3_stmt* s = nullptr;
-  int r = sqlite3_prepare_v2(db->connection_, *sql, -1, &s, nullptr);
+
+  unsigned int prep_flags =
+      persistent.value_or(false) ? SQLITE_PREPARE_PERSISTENT : 0;
+
+  int r =
+      sqlite3_prepare_v3(db->connection_, *sql, -1, prep_flags, &s, nullptr);
+
   StatementPtr stmt_ptr(s);
 
   CHECK_ERROR_OR_THROW(env->isolate(), db, r, SQLITE_OK, void());
@@ -1636,6 +1689,7 @@ void DatabaseSync::Exec(const FunctionCallbackInfo<Value>& args) {
   ASSIGN_OR_RETURN_UNWRAP(&db, args.This());
   Environment* env = Environment::GetCurrent(args);
   THROW_AND_RETURN_ON_BAD_STATE(env, !db->IsOpen(), "database is not open");
+  THROW_AND_RETURN_IF_IN_AUTHORIZER(env, db);
 
   if (!args[0]->IsString()) {
     THROW_ERR_INVALID_ARG_TYPE(env->isolate(),
@@ -1660,6 +1714,7 @@ void DatabaseSync::CustomFunction(const FunctionCallbackInfo<Value>& args) {
   ASSIGN_OR_RETURN_UNWRAP(&db, args.This());
   Environment* env = Environment::GetCurrent(args);
   THROW_AND_RETURN_ON_BAD_STATE(env, !db->IsOpen(), "database is not open");
+  THROW_AND_RETURN_IF_IN_AUTHORIZER(env, db);
 
   if (!args[0]->IsString()) {
     THROW_ERR_INVALID_ARG_TYPE(env->isolate(),
@@ -1833,6 +1888,7 @@ void DatabaseSync::Serialize(const FunctionCallbackInfo<Value>& args) {
   ASSIGN_OR_RETURN_UNWRAP(&db, args.This());
   Environment* env = Environment::GetCurrent(args);
   THROW_AND_RETURN_ON_BAD_STATE(env, !db->IsOpen(), "database is not open");
+  THROW_AND_RETURN_IF_IN_AUTHORIZER(env, db);
 
   std::string db_name = "main";
   if (!args[0]->IsUndefined()) {
@@ -1967,6 +2023,7 @@ void DatabaseSync::AggregateFunction(const FunctionCallbackInfo<Value>& args) {
   ASSIGN_OR_RETURN_UNWRAP(&db, args.This());
   Environment* env = Environment::GetCurrent(args);
   THROW_AND_RETURN_ON_BAD_STATE(env, !db->IsOpen(), "database is not open");
+  THROW_AND_RETURN_IF_IN_AUTHORIZER(env, db);
   Utf8Value name(env->isolate(), args[0].As<String>());
   Local<Object> options = args[1].As<Object>();
   Local<Value> start_v;
@@ -2178,6 +2235,7 @@ void DatabaseSync::CreateSession(const FunctionCallbackInfo<Value>& args) {
   DatabaseSync* db;
   ASSIGN_OR_RETURN_UNWRAP(&db, args.This());
   THROW_AND_RETURN_ON_BAD_STATE(env, !db->IsOpen(), "database is not open");
+  THROW_AND_RETURN_IF_IN_AUTHORIZER(env, db);
 
   sqlite3_session* pSession;
   int r = sqlite3session_create(db->connection_, db_name.c_str(), &pSession);
@@ -2347,6 +2405,7 @@ void DatabaseSync::ApplyChangeset(const FunctionCallbackInfo<Value>& args) {
   ASSIGN_OR_RETURN_UNWRAP(&db, args.This());
   Environment* env = Environment::GetCurrent(args);
   THROW_AND_RETURN_ON_BAD_STATE(env, !db->IsOpen(), "database is not open");
+  THROW_AND_RETURN_IF_IN_AUTHORIZER(env, db);
 
   if (!args[0]->IsUint8Array()) {
     THROW_ERR_INVALID_ARG_TYPE(
@@ -2482,6 +2541,7 @@ void DatabaseSync::EnableLoadExtension(
   ASSIGN_OR_RETURN_UNWRAP(&db, args.This());
   Environment* env = Environment::GetCurrent(args);
   THROW_AND_RETURN_ON_BAD_STATE(env, !db->IsOpen(), "database is not open");
+  THROW_AND_RETURN_IF_IN_AUTHORIZER(env, db);
 
   Isolate* isolate = env->isolate();
   if (!args[0]->IsBoolean()) {
@@ -2510,6 +2570,7 @@ void DatabaseSync::EnableDefensive(const FunctionCallbackInfo<Value>& args) {
   ASSIGN_OR_RETURN_UNWRAP(&db, args.This());
   Environment* env = Environment::GetCurrent(args);
   THROW_AND_RETURN_ON_BAD_STATE(env, !db->IsOpen(), "database is not open");
+  THROW_AND_RETURN_IF_IN_AUTHORIZER(env, db);
 
   Isolate* isolate = env->isolate();
   if (!args[0]->IsBoolean()) {
@@ -2535,6 +2596,7 @@ void DatabaseSync::LoadExtension(const FunctionCallbackInfo<Value>& args) {
       env, !db->allow_load_extension_, "extension loading is not allowed");
   THROW_AND_RETURN_ON_BAD_STATE(
       env, !db->enable_load_extension_, "extension loading is not allowed");
+  THROW_AND_RETURN_IF_IN_AUTHORIZER(env, db);
 
   if (!args[0]->IsString()) {
     THROW_ERR_INVALID_ARG_TYPE(env->isolate(),
@@ -2563,6 +2625,7 @@ void DatabaseSync::SetAuthorizer(const FunctionCallbackInfo<Value>& args) {
   ASSIGN_OR_RETURN_UNWRAP(&db, args.This());
   Environment* env = Environment::GetCurrent(args);
   THROW_AND_RETURN_ON_BAD_STATE(env, !db->IsOpen(), "database is not open");
+  THROW_AND_RETURN_IF_IN_AUTHORIZER(env, db);
 
   Isolate* isolate = env->isolate();
 
@@ -2599,6 +2662,7 @@ int DatabaseSync::AuthorizerCallback(void* user_data,
                                      const char* param4) {
   DatabaseSync* db = static_cast<DatabaseSync*>(user_data);
   CallbackDepthGuard guard(db);
+  AuthorizerDepthGuard authorizer_guard(db);
   Environment* env = db->env();
   Isolate* isolate = env->isolate();
   HandleScope handle_scope(isolate);
@@ -2712,12 +2776,20 @@ void StatementSync::Close(const FunctionCallbackInfo<Value>& args) {
   Environment* env = Environment::GetCurrent(args);
   THROW_AND_RETURN_ON_BAD_STATE(
       env, stmt->IsFinalized(), "statement has been finalized");
+  THROW_AND_RETURN_IF_STEPPING(env, stmt);
   stmt->Close();
 }
 
 void StatementSync::Dispose(const FunctionCallbackInfo<Value>& args) {
   StatementSync* stmt;
   ASSIGN_OR_RETURN_UNWRAP(&stmt, args.This());
+  Environment* env = Environment::GetCurrent(args);
+  // Disposal is idempotent, so an already-finalized statement is a no-op even
+  // inside a callback.
+  if (stmt->IsFinalized()) {
+    return;
+  }
+  THROW_AND_RETURN_IF_STEPPING(env, stmt);
   stmt->Close();
 }
 
@@ -2999,6 +3071,7 @@ MaybeLocal<Value> StatementExecutionHelper::All(Environment* env,
   LocalVector<Value> row_values(isolate);
   LocalVector<Name> row_keys(isolate);
 
+  SteppingStatementGuard stepping(db, stmt);
   while ((r = sqlite3_step(stmt)) == SQLITE_ROW) {
     if (num_cols == 0) {
       num_cols = sqlite3_column_count(stmt);
@@ -3041,6 +3114,9 @@ MaybeLocal<Object> StatementExecutionHelper::Run(Environment* env,
                                                  bool use_big_ints) {
   Isolate* isolate = env->isolate();
   EscapableHandleScope scope(isolate);
+  // Declared before the reset below so that it outlives it: sqlite3_reset()
+  // can run JavaScript through an aggregate's xFinal.
+  SteppingStatementGuard stepping(db, stmt);
   bool needs_reset = true;
   auto reset = OnScopeLeave([&]() {
     if (needs_reset) sqlite3_reset(stmt);
@@ -3128,6 +3204,9 @@ MaybeLocal<Value> StatementExecutionHelper::Get(Environment* env,
                                                 bool use_big_ints) {
   Isolate* isolate = env->isolate();
   EscapableHandleScope scope(isolate);
+  // Declared before the reset below so that it outlives it: sqlite3_reset()
+  // can run JavaScript through an aggregate's xFinal.
+  SteppingStatementGuard stepping(db, stmt);
   bool needs_reset = true;
   auto reset = OnScopeLeave([&]() {
     if (needs_reset) sqlite3_reset(stmt);
@@ -3184,6 +3263,8 @@ void StatementSync::All(const FunctionCallbackInfo<Value>& args) {
   Environment* env = Environment::GetCurrent(args);
   THROW_AND_RETURN_ON_BAD_STATE(
       env, stmt->IsFinalized(), "statement has been finalized");
+  THROW_AND_RETURN_IF_IN_AUTHORIZER(env, stmt->db_.get());
+  THROW_AND_RETURN_IF_STEPPING(env, stmt);
   Isolate* isolate = env->isolate();
   int r = stmt->ResetStatement();
   CHECK_ERROR_OR_THROW(isolate, stmt->db_.get(), r, SQLITE_OK, void());
@@ -3215,6 +3296,8 @@ void StatementSync::Iterate(const FunctionCallbackInfo<Value>& args) {
   Environment* env = Environment::GetCurrent(args);
   THROW_AND_RETURN_ON_BAD_STATE(
       env, stmt->IsFinalized(), "statement has been finalized");
+  THROW_AND_RETURN_IF_IN_AUTHORIZER(env, stmt->db_.get());
+  THROW_AND_RETURN_IF_STEPPING(env, stmt);
   int r = stmt->ResetStatement();
   CHECK_ERROR_OR_THROW(env->isolate(), stmt->db_.get(), r, SQLITE_OK, void());
 
@@ -3238,6 +3321,8 @@ void StatementSync::Get(const FunctionCallbackInfo<Value>& args) {
   Environment* env = Environment::GetCurrent(args);
   THROW_AND_RETURN_ON_BAD_STATE(
       env, stmt->IsFinalized(), "statement has been finalized");
+  THROW_AND_RETURN_IF_IN_AUTHORIZER(env, stmt->db_.get());
+  THROW_AND_RETURN_IF_STEPPING(env, stmt);
   int r = stmt->ResetStatement();
   CHECK_ERROR_OR_THROW(env->isolate(), stmt->db_.get(), r, SQLITE_OK, void());
 
@@ -3262,6 +3347,8 @@ void StatementSync::Run(const FunctionCallbackInfo<Value>& args) {
   Environment* env = Environment::GetCurrent(args);
   THROW_AND_RETURN_ON_BAD_STATE(
       env, stmt->IsFinalized(), "statement has been finalized");
+  THROW_AND_RETURN_IF_IN_AUTHORIZER(env, stmt->db_.get());
+  THROW_AND_RETURN_IF_STEPPING(env, stmt);
   int r = stmt->ResetStatement();
   CHECK_ERROR_OR_THROW(env->isolate(), stmt->db_.get(), r, SQLITE_OK, void());
 
@@ -3598,6 +3685,7 @@ void SQLTagStore::Run(const FunctionCallbackInfo<Value>& args) {
 
   THROW_AND_RETURN_ON_BAD_STATE(
       env, !session->database_->IsOpen(), "database is not open");
+  THROW_AND_RETURN_IF_IN_AUTHORIZER(env, session->database_.get());
 
   BaseObjectPtr<StatementSync> stmt = PrepareStatement(args);
 
@@ -3605,6 +3693,8 @@ void SQLTagStore::Run(const FunctionCallbackInfo<Value>& args) {
     return;
   }
 
+  THROW_AND_RETURN_IF_STEPPING(env, stmt.get());
+
   if (!ResetAndBindStatement(env, stmt.get(), args)) {
     return;
   }
@@ -3624,6 +3714,7 @@ void SQLTagStore::Iterate(const FunctionCallbackInfo<Value>& args) {
 
   THROW_AND_RETURN_ON_BAD_STATE(
       env, !session->database_->IsOpen(), "database is not open");
+  THROW_AND_RETURN_IF_IN_AUTHORIZER(env, session->database_.get());
 
   BaseObjectPtr<StatementSync> stmt = PrepareStatement(args);
 
@@ -3631,6 +3722,8 @@ void SQLTagStore::Iterate(const FunctionCallbackInfo<Value>& args) {
     return;
   }
 
+  THROW_AND_RETURN_IF_STEPPING(env, stmt.get());
+
   if (!ResetAndBindStatement(env, stmt.get(), args)) {
     return;
   }
@@ -3652,6 +3745,7 @@ void SQLTagStore::Get(const FunctionCallbackInfo<Value>& args) {
 
   THROW_AND_RETURN_ON_BAD_STATE(
       env, !session->database_->IsOpen(), "database is not open");
+  THROW_AND_RETURN_IF_IN_AUTHORIZER(env, session->database_.get());
 
   BaseObjectPtr<StatementSync> stmt = PrepareStatement(args);
 
@@ -3659,6 +3753,8 @@ void SQLTagStore::Get(const FunctionCallbackInfo<Value>& args) {
     return;
   }
 
+  THROW_AND_RETURN_IF_STEPPING(env, stmt.get());
+
   if (!ResetAndBindStatement(env, stmt.get(), args)) {
     return;
   }
@@ -3681,6 +3777,7 @@ void SQLTagStore::All(const FunctionCallbackInfo<Value>& args) {
 
   THROW_AND_RETURN_ON_BAD_STATE(
       env, !session->database_->IsOpen(), "database is not open");
+  THROW_AND_RETURN_IF_IN_AUTHORIZER(env, session->database_.get());
 
   BaseObjectPtr<StatementSync> stmt = PrepareStatement(args);
 
@@ -3688,6 +3785,8 @@ void SQLTagStore::All(const FunctionCallbackInfo<Value>& args) {
     return;
   }
 
+  THROW_AND_RETURN_IF_STEPPING(env, stmt.get());
+
   if (!ResetAndBindStatement(env, stmt.get(), args)) {
     return;
   }
@@ -3713,6 +3812,10 @@ void SQLTagStore::All(const FunctionCallbackInfo<Value>& args) {
 void SQLTagStore::Clear(const FunctionCallbackInfo<Value>& args) {
   SQLTagStore* store;
   ASSIGN_OR_RETURN_UNWRAP(&store, args.This());
+  // Clearing the cache drops strong references to the cached statements but
+  // never finalizes one synchronously, so it touches no SQLite state and stays
+  // available from a callback. Invalidating the cache after a schema change is
+  // a legitimate use of an authorizer.
   store->sql_tags_.Clear();
 }
 
@@ -3766,8 +3869,12 @@ BaseObjectPtr<StatementSync> SQLTagStore::PrepareStatement(
 
   if (stmt == nullptr) {
     sqlite3_stmt* s = nullptr;
-    int r = sqlite3_prepare_v2(
-        session->database_->connection_, sql.data(), sql.size(), &s, nullptr);
+    int r = sqlite3_prepare_v3(session->database_->connection_,
+                               sql.data(),
+                               sql.size(),
+                               SQLITE_PREPARE_PERSISTENT,
+                               &s,
+                               nullptr);
     StatementPtr stmt_ptr(s);
 
     if (r != SQLITE_OK) {
@@ -3922,6 +4029,8 @@ void StatementSyncIterator::Next(const FunctionCallbackInfo<Value>& args) {
 
   auto iter_template = getLazyIterTemplate(env);
 
+  // A drained iterator touches no SQLite state, so it stays usable from a
+  // callback and is checked before the guards below.
   if (iter->done_) {
     MaybeLocal<Value> values[]{
         Boolean::New(isolate, true),
@@ -3935,11 +4044,18 @@ void StatementSyncIterator::Next(const FunctionCallbackInfo<Value>& args) {
     return;
   }
 
+  THROW_AND_RETURN_IF_IN_AUTHORIZER(env, iter->stmt_->db_.get());
+  THROW_AND_RETURN_IF_STEPPING(env, iter->stmt_.get());
+
   THROW_AND_RETURN_ON_BAD_STATE(
       env,
       iter->statement_reset_generation_ != iter->stmt_->reset_generation_,
       "iterator was invalidated");
 
+  // sqlite3_reset() can run JavaScript through an aggregate's xFinal, so it
+  // stays inside the guard.
+  SteppingStatementGuard stepping(iter->stmt_->db_.get(),
+                                  iter->stmt_->statement_.get());
   int r = sqlite3_step(iter->stmt_->statement_.get());
   if (r != SQLITE_ROW) {
     CHECK_ERROR_OR_THROW(
@@ -4000,11 +4116,18 @@ void StatementSyncIterator::Return(const FunctionCallbackInfo<Value>& args) {
       env, iter->stmt_->IsFinalized(), "statement has been finalized");
   Isolate* isolate = env->isolate();
 
-  // Unlike Next(), the reset result is intentionally ignored here: Return()
-  // is invoked by the language during abrupt completion (e.g. a `throw`
-  // inside a `for...of` body), and throwing on a deferred SQLite error
-  // would discard the caller's already-pending exception.
-  sqlite3_reset(iter->stmt_->statement_.get());
+  if (!iter->done_) {
+    // A language-invoked return() cannot be reached mid-step, since the loop
+    // body only runs after next() has returned, so these guards only reject an
+    // explicit call from inside a callback.
+    THROW_AND_RETURN_IF_IN_AUTHORIZER(env, iter->stmt_->db_.get());
+    THROW_AND_RETURN_IF_STEPPING(env, iter->stmt_.get());
+    // Unlike Next(), the reset result is intentionally ignored here: Return()
+    // is invoked by the language during abrupt completion (e.g. a `throw`
+    // inside a `for...of` body), and throwing on a deferred SQLite error
+    // would discard the caller's already-pending exception.
+    sqlite3_reset(iter->stmt_->statement_.get());
+  }
   iter->done_ = true;
 
   auto iter_template = getLazyIterTemplate(env);
@@ -4080,6 +4203,7 @@ void Session::Changeset(const FunctionCallbackInfo<Value>& args) {
       env, !session->database_->IsOpen(), "database is not open");
   THROW_AND_RETURN_ON_BAD_STATE(
       env, session->session_ == nullptr, "session is not open");
+  THROW_AND_RETURN_IF_IN_AUTHORIZER(env, session->database_.get());
 
   int nChangeset;
   void* pChangeset;
diff --git a/src/node_sqlite.h b/src/node_sqlite.h
index 5f51a48fe1cc..daa8d5b8a37e 100644
--- a/src/node_sqlite.h
+++ b/src/node_sqlite.h
@@ -9,6 +9,7 @@
 #include "sqlite3.h"
 #include "util.h"
 
+#include <algorithm>
 #include <array>
 #include <list>
 #include <map>
@@ -265,6 +266,26 @@ class DatabaseSync : public BaseObject {
   void DecrementCallbackDepth() { --callback_depth_; }
   bool IsInCallback() const { return callback_depth_ > 0; }
 
+  // SQLite forbids an authorizer callback from doing anything that modifies
+  // the database connection that invoked it, which includes preparing and
+  // stepping statements. See https://www.sqlite.org/c3ref/set_authorizer.html.
+  void IncrementAuthorizerDepth() { ++authorizer_depth_; }
+  void DecrementAuthorizerDepth() { --authorizer_depth_; }
+  bool IsInAuthorizerCallback() const { return authorizer_depth_ > 0; }
+
+  // Finalizing a statement frees its virtual machine, so a callback that
+  // SQLite invokes from inside sqlite3_step() must not finalize the statement
+  // being stepped. Other statements on the connection are safe to finalize.
+  void PushSteppingStatement(sqlite3_stmt* stmt) {
+    stepping_statements_.push_back(stmt);
+  }
+  void PopSteppingStatement() { stepping_statements_.pop_back(); }
+  bool IsSteppingStatement(sqlite3_stmt* stmt) const {
+    return std::find(stepping_statements_.begin(),
+                     stepping_statements_.end(),
+                     stmt) != stepping_statements_.end();
+  }
+
   SET_MEMORY_INFO_NAME(DatabaseSync)
   SET_SELF_SIZE(DatabaseSync)
 
@@ -279,6 +300,8 @@ class DatabaseSync : public BaseObject {
   sqlite3* connection_;
   bool ignore_next_sqlite_error_;
   int callback_depth_ = 0;
+  int authorizer_depth_ = 0;
+  std::vector<sqlite3_stmt*> stepping_statements_;
 
   std::set<BackupJob*> backups_;
   std::unordered_set<Session*> sessions_;
@@ -460,6 +483,32 @@ class CallbackDepthGuard {
   DatabaseSync* db_;
 };
 
+class SteppingStatementGuard {
+ public:
+  SteppingStatementGuard(DatabaseSync* db, sqlite3_stmt* stmt) : db_(db) {
+    db_->PushSteppingStatement(stmt);
+  }
+  ~SteppingStatementGuard() { db_->PopSteppingStatement(); }
+  SteppingStatementGuard(const SteppingStatementGuard&) = delete;
+  SteppingStatementGuard& operator=(const SteppingStatementGuard&) = delete;
+
+ private:
+  DatabaseSync* db_;
+};
+
+class AuthorizerDepthGuard {
+ public:
+  explicit AuthorizerDepthGuard(DatabaseSync* db) : db_(db) {
+    db_->IncrementAuthorizerDepth();
+  }
+  ~AuthorizerDepthGuard() { db_->DecrementAuthorizerDepth(); }
+  AuthorizerDepthGuard(const AuthorizerDepthGuard&) = delete;
+  AuthorizerDepthGuard& operator=(const AuthorizerDepthGuard&) = delete;
+
+ private:
+  DatabaseSync* db_;
+};
+
 class UserDefinedFunction {
  public:
   UserDefinedFunction(Environment* env,
diff --git a/test/parallel/test-sqlite-authz.js b/test/parallel/test-sqlite-authz.js
index 69c075a57e2e..f6020ce9047e 100644
--- a/test/parallel/test-sqlite-authz.js
+++ b/test/parallel/test-sqlite-authz.js
@@ -1,7 +1,7 @@
 'use strict';
 
-const { skipIfSQLiteMissing } = require('../common');
-skipIfSQLiteMissing();
+const common = require('../common');
+common.skipIfSQLiteMissing();
 
 const assert = require('node:assert');
 const { DatabaseSync, constants } = require('node:sqlite');
@@ -288,3 +288,329 @@ suite('DatabaseSync.prototype.setAuthorizer()', () => {
     });
   });
 });
+
+// SQLite forbids an authorizer callback from modifying the connection that
+// invoked it, which includes preparing and stepping statements.
+// See https://www.sqlite.org/c3ref/set_authorizer.html.
+suite('authorizer callback reentrancy', () => {
+  const expectedError = 'ERR_INVALID_STATE: database cannot be accessed ' +
+    'from an authorizer callback';
+  const steppingError =
+    'ERR_INVALID_STATE: statement is already being executed';
+
+  // Calls each of `cases` from inside an authorizer callback, and returns a
+  // `name -> outcome` map of what each one threw.
+  const runInAuthorizer = (db, cases) => {
+    const outcomes = {};
+    for (const [name, fn] of Object.entries(cases)) {
+      let ran = false;
+      db.setAuthorizer(() => {
+        if (!ran) {
+          ran = true;
+          try {
+            fn();
+            outcomes[name] = 'did not throw';
+          } catch (err) {
+            outcomes[name] = `${err.code}: ${err.message}`;
+          }
+        }
+        return constants.SQLITE_OK;
+      });
+      db.exec('SELECT 1');
+      db.setAuthorizer(null);
+      if (!ran) {
+        outcomes[name] = 'authorizer callback did not run';
+      }
+    }
+    return outcomes;
+  };
+
+  // Builds the expected `name -> outcome` map for the given case names.
+  const allRejected = (cases) => Object.fromEntries(
+    Object.keys(cases).map((name) => [name, expectedError]),
+  );
+
+  it('rejects database methods', () => {
+    const db = new DatabaseSync(':memory:');
+    db.exec('CREATE TABLE t (x INTEGER)');
+    const cases = {
+      prepare: () => db.prepare('SELECT 1'),
+      exec: () => db.exec('SELECT 1'),
+      setAuthorizer: () => db.setAuthorizer(null),
+      createSession: () => db.createSession(),
+      applyChangeset: () => db.applyChangeset(new Uint8Array([1])),
+      createTagStore: () => db.createTagStore(),
+      serialize: () => db.serialize(),
+      function: () => db.function('noop', () => 1),
+      aggregate: () => db.aggregate('agg', { start: 0, step: (acc) => acc }),
+      enableLoadExtension: () => db.enableLoadExtension(false),
+      enableDefensive: () => db.enableDefensive(true),
+      limits: () => { db.limits.length = 100; },
+    };
+
+    assert.deepStrictEqual(runInAuthorizer(db, cases), allRejected(cases));
+  });
+
+  // loadExtension() checks that extension loading is enabled before reaching
+  // the authorizer guard, so it needs a database opened with allowExtension.
+  it('rejects loadExtension', () => {
+    const db = new DatabaseSync(':memory:', { allowExtension: true });
+    db.enableLoadExtension(true);
+    db.exec('CREATE TABLE t (x INTEGER)');
+    const cases = {
+      loadExtension: () => db.loadExtension('/nonexistent/extension'),
+    };
+
+    assert.deepStrictEqual(runInAuthorizer(db, cases), allRejected(cases));
+  });
+
+  // close() and deserialize() tear down the connection, so the pre-existing
+  // callback depth guard already rejects them with its own message.
+  it('rejects methods the callback depth guard already covers', () => {
+    const db = new DatabaseSync(':memory:');
+    db.exec('CREATE TABLE t (x INTEGER)');
+    const snapshot = db.serialize();
+    const cases = {
+      close: () => db.close(),
+      deserialize: () => db.deserialize(snapshot),
+    };
+
+    assert.deepStrictEqual(runInAuthorizer(db, cases), {
+      close: 'ERR_INVALID_STATE: database cannot be closed while in a callback',
+      deserialize: 'ERR_INVALID_STATE: database cannot be deserialized ' +
+        'while in a callback',
+    });
+  });
+
+  it('rejects statement methods', () => {
+    const db = new DatabaseSync(':memory:');
+    db.exec('CREATE TABLE t (x INTEGER)');
+    db.exec('INSERT INTO t VALUES (1)');
+    const stmt = db.prepare('SELECT x FROM t');
+    const cases = {
+      run: () => stmt.run(),
+      get: () => stmt.get(),
+      all: () => stmt.all(),
+      iterate: () => stmt.iterate(),
+    };
+
+    assert.deepStrictEqual(runInAuthorizer(db, cases), allRejected(cases));
+  });
+
+  // Only the statement being stepped is unsafe to finalize. Other statements
+  // on the connection have their own virtual machines, so finalizing them from
+  // a callback is allowed.
+  it('allows finalizing a statement that is not being executed', () => {
+    const db = new DatabaseSync(':memory:');
+    db.exec('CREATE TABLE t (x INTEGER)');
+    db.exec('INSERT INTO t VALUES (1)');
+    const closeStmt = db.prepare('SELECT x FROM t');
+    const disposeStmt = db.prepare('SELECT x FROM t');
+    const cases = {
+      close: () => closeStmt.close(),
+      dispose: () => disposeStmt[Symbol.dispose](),
+    };
+
+    assert.deepStrictEqual(runInAuthorizer(db, cases), {
+      close: 'did not throw',
+      dispose: 'did not throw',
+    });
+  });
+
+  // Disposal is idempotent, so a statement that is already finalized must stay
+  // a no-op even inside a callback. Throwing here would turn a `using` scope's
+  // real exception into a SuppressedError.
+  it('allows disposing an already-finalized statement', () => {
+    const db = new DatabaseSync(':memory:');
+    db.exec('CREATE TABLE t (x INTEGER)');
+    const stmt = db.prepare('SELECT x FROM t');
+    stmt.close();
+    const cases = { dispose: () => stmt[Symbol.dispose]() };
+
+    assert.deepStrictEqual(runInAuthorizer(db, cases), {
+      dispose: 'did not throw',
+    });
+  });
+
+  it('rejects session changeset methods', () => {
+    const db = new DatabaseSync(':memory:');
+    db.exec('CREATE TABLE t (x INTEGER PRIMARY KEY, y TEXT)');
+    const session = db.createSession({ table: 't' });
+    db.exec("INSERT INTO t VALUES (1, 'a')");
+    const cases = {
+      changeset: () => session.changeset(),
+      patchset: () => session.patchset(),
+    };
+
+    assert.deepStrictEqual(runInAuthorizer(db, cases), allRejected(cases));
+  });
+
+  // A statement being re-prepared inside sqlite3_step() is the case that
+  // actually crashes, because that statement's VM is mid-execution.
+  it('rejects finalizing the statement being stepped', () => {
+    const db = new DatabaseSync(':memory:');
+    db.exec('CREATE TABLE t (x INTEGER)');
+    db.exec('INSERT INTO t VALUES (1)');
+    const stmt = db.prepare('SELECT x FROM t');
+    stmt.get();
+    db.exec('ALTER TABLE t ADD COLUMN y INTEGER');
+
+    let outcome = 'authorizer callback did not run';
+    let ran = false;
+    db.setAuthorizer(() => {
+      if (!ran) {
+        ran = true;
+        try {
+          stmt.close();
+          outcome = 'did not throw';
+        } catch (err) {
+          outcome = `${err.code}: ${err.message}`;
+        }
+      }
+      return constants.SQLITE_OK;
+    });
+
+    stmt.get();
+
+    assert.strictEqual(outcome, steppingError);
+  });
+
+  // Unlike an already-finalized statement, disposing the one being stepped
+  // would free the running virtual machine, so it throws.
+  it('rejects disposing the statement being stepped', () => {
+    const db = new DatabaseSync(':memory:');
+    db.exec('CREATE TABLE t (x INTEGER)');
+    db.exec('INSERT INTO t VALUES (1)');
+    const stmt = db.prepare('SELECT x FROM t');
+    stmt.get();
+    db.exec('ALTER TABLE t ADD COLUMN y INTEGER');
+
+    let outcome = 'authorizer callback did not run';
+    let ran = false;
+    db.setAuthorizer(() => {
+      if (!ran) {
+        ran = true;
+        try {
+          stmt[Symbol.dispose]();
+          outcome = 'did not throw';
+        } catch (err) {
+          outcome = `${err.code}: ${err.message}`;
+        }
+      }
+      return constants.SQLITE_OK;
+    });
+
+    stmt.get();
+
+    assert.strictEqual(outcome, steppingError);
+  });
+
+  it('rejects iterator methods', () => {
+    const db = new DatabaseSync(':memory:');
+    db.exec('CREATE TABLE t (x INTEGER)');
+    db.exec('INSERT INTO t VALUES (1), (2)');
+    const iter = db.prepare('SELECT x FROM t').iterate();
+    const cases = {
+      next: () => iter.next(),
+      return: () => iter.return(),
+    };
+
+    assert.deepStrictEqual(runInAuthorizer(db, cases), allRejected(cases));
+    iter.return();
+  });
+
+  // A drained iterator holds no SQLite state, so next() and return() stay
+  // available and remain idempotent inside a callback.
+  it('allows iterator methods on a drained iterator', () => {
+    const db = new DatabaseSync(':memory:');
+    db.exec('CREATE TABLE t (x INTEGER)');
+    db.exec('INSERT INTO t VALUES (1)');
+    const iter = db.prepare('SELECT x FROM t').iterate();
+    for (const row of iter) {
+      assert.ok(row);
+    }
+    const done = {};
+    const cases = {
+      next: () => { done.next = iter.next().done; },
+      return: () => { done.return = iter.return().done; },
+    };
+
+    assert.deepStrictEqual(runInAuthorizer(db, cases), {
+      next: 'did not throw',
+      return: 'did not throw',
+    });
+    assert.deepStrictEqual(done, { next: true, return: true });
+  });
+
+  it('rejects tag store methods', () => {
+    const db = new DatabaseSync(':memory:');
+    db.exec('CREATE TABLE t (x INTEGER)');
+    const sql = db.createTagStore(10);
+    const cases = {
+      run: () => sql.run`SELECT 1`,
+      get: () => sql.get`SELECT 1`,
+      all: () => sql.all`SELECT 1`,
+      iterate: () => sql.iterate`SELECT 1`,
+    };
+
+    assert.deepStrictEqual(runInAuthorizer(db, cases), allRejected(cases));
+  });
+
+  // clear() only drops cached statements, so invalidating the cache after a
+  // schema change is allowed from the callback.
+  it('allows clearing a tag store', () => {
+    const db = new DatabaseSync(':memory:');
+    db.exec('CREATE TABLE t (x INTEGER)');
+    db.exec('INSERT INTO t VALUES (1)');
+    const sql = db.createTagStore(10);
+    assert.strictEqual(sql.all`SELECT x FROM t`.length, 1);
+    assert.strictEqual(sql.size, 1);
+    const cases = { clear: () => sql.clear() };
+
+    assert.deepStrictEqual(runInAuthorizer(db, cases), {
+      clear: 'did not throw',
+    });
+    assert.strictEqual(sql.size, 0);
+  });
+
+  // A statement may be re-prepared during sqlite3_step() after a schema
+  // change, which invokes the authorizer without an explicit prepare() call.
+  it('rejects reentry when the authorizer runs during a re-prepare', () => {
+    const db = new DatabaseSync(':memory:');
+    db.exec('CREATE TABLE t (x INTEGER)');
+    db.exec('INSERT INTO t VALUES (1)');
+    const stmt = db.prepare('SELECT x FROM t');
+    stmt.get();
+    db.exec('ALTER TABLE t ADD COLUMN y INTEGER');
+
+    let outcome = 'authorizer callback did not run';
+    let ran = false;
+    db.setAuthorizer(() => {
+      if (!ran) {
+        ran = true;
+        try {
+          db.prepare('SELECT 1');
+          outcome = 'did not throw';
+        } catch (err) {
+          outcome = `${err.code}: ${err.message}`;
+        }
+      }
+      return constants.SQLITE_OK;
+    });
+
+    stmt.get();
+
+    assert.strictEqual(outcome, expectedError);
+  });
+
+  it('allows access again after the authorizer returns', () => {
+    const db = new DatabaseSync(':memory:');
+    db.exec('CREATE TABLE t (x INTEGER)');
+    const cases = { prepare: () => db.prepare('SELECT 1') };
+
+    assert.deepStrictEqual(runInAuthorizer(db, cases), allRejected(cases));
+
+    db.setAuthorizer(() => constants.SQLITE_OK);
+    assert.deepStrictEqual(db.prepare('SELECT 1 AS v').get(), { __proto__: null, v: 1 });
+  });
+});
diff --git a/test/parallel/test-sqlite-statement-sync.js b/test/parallel/test-sqlite-statement-sync.js
index a52634f59ac6..b13da41e9d71 100644
--- a/test/parallel/test-sqlite-statement-sync.js
+++ b/test/parallel/test-sqlite-statement-sync.js
@@ -18,8 +18,7 @@ suite('StatementSync() constructor', () => {
 
 suite('StatementSync.prototype.get()', () => {
   test('executes a query and returns undefined on no results', (t) => {
-    const db = new DatabaseSync(':memory:');
-    t.after(() => { db.close(); });
+    using db = new DatabaseSync(':memory:');
     let stmt = db.prepare('CREATE TABLE storage(key TEXT, val TEXT)');
     t.assert.strictEqual(stmt.get(), undefined);
     stmt = db.prepare('SELECT * FROM storage');
@@ -27,8 +26,7 @@ suite('StatementSync.prototype.get()', () => {
   });
 
   test('executes a query and returns the first result', (t) => {
-    const db = new DatabaseSync(':memory:');
-    t.after(() => { db.close(); });
+    using db = new DatabaseSync(':memory:');
     let stmt = db.prepare('CREATE TABLE storage(key TEXT, val TEXT)');
     t.assert.strictEqual(stmt.get(), undefined);
     stmt = db.prepare('INSERT INTO storage (key, val) VALUES (?, ?)');
@@ -39,8 +37,7 @@ suite('StatementSync.prototype.get()', () => {
   });
 
   test('executes a query that returns special columns', (t) => {
-    const db = new DatabaseSync(':memory:');
-    t.after(() => { db.close(); });
+    using db = new DatabaseSync(':memory:');
     const stmt = db.prepare('SELECT 1 as __proto__, 2 as constructor, 3 as toString');
     t.assert.deepStrictEqual(stmt.get(), { __proto__: null, ['__proto__']: 1, constructor: 2, toString: 3 });
   });
@@ -105,15 +102,13 @@ suite('StatementSync.prototype.get()', () => {
 
 suite('StatementSync.prototype.all()', () => {
   test('executes a query and returns an empty array on no results', (t) => {
-    const db = new DatabaseSync(':memory:');
-    t.after(() => { db.close(); });
+    using db = new DatabaseSync(':memory:');
     const stmt = db.prepare('CREATE TABLE storage(key TEXT, val TEXT)');
     t.assert.deepStrictEqual(stmt.all(), []);
   });
 
   test('executes a query and returns all results', (t) => {
-    const db = new DatabaseSync(':memory:');
-    t.after(() => { db.close(); });
+    using db = new DatabaseSync(':memory:');
     let stmt = db.prepare('CREATE TABLE storage(key TEXT, val TEXT)');
     t.assert.deepStrictEqual(stmt.run(), { changes: 0, lastInsertRowid: 0 });
     stmt = db.prepare('INSERT INTO storage (key, val) VALUES (?, ?)');
@@ -189,8 +184,7 @@ suite('StatementSync.prototype.all()', () => {
 
 suite('StatementSync.prototype.iterate()', () => {
   test('executes a query and returns an empty iterator on no results', (t) => {
-    const db = new DatabaseSync(':memory:');
-    t.after(() => { db.close(); });
+    using db = new DatabaseSync(':memory:');
     const stmt = db.prepare('CREATE TABLE storage(key TEXT, val TEXT)');
     const iter = stmt.iterate();
     t.assert.strictEqual(iter instanceof globalThis.Iterator, true);
@@ -199,8 +193,7 @@ suite('StatementSync.prototype.iterate()', () => {
   });
 
   test('executes a query and returns all results', (t) => {
-    const db = new DatabaseSync(':memory:');
-    t.after(() => { db.close(); });
+    using db = new DatabaseSync(':memory:');
     let stmt = db.prepare('CREATE TABLE storage(key TEXT, val TEXT)');
     t.assert.deepStrictEqual(stmt.run(), { changes: 0, lastInsertRowid: 0 });
     stmt = db.prepare('INSERT INTO storage (key, val) VALUES (?, ?)');
@@ -402,8 +395,7 @@ suite('StatementSync.prototype.iterate()', () => {
 
 suite('StatementSync.prototype.run()', () => {
   test('executes a query and returns change metadata', (t) => {
-    const db = new DatabaseSync(':memory:');
-    t.after(() => { db.close(); });
+    using db = new DatabaseSync(':memory:');
     const setup = db.exec(`
       CREATE TABLE storage(key TEXT, val TEXT);
       INSERT INTO storage (key, val) VALUES ('foo', 'bar');
@@ -414,8 +406,7 @@ suite('StatementSync.prototype.run()', () => {
   });
 
   test('SQLite throws when trying to bind too many parameters', (t) => {
-    const db = new DatabaseSync(':memory:');
-    t.after(() => { db.close(); });
+    using db = new DatabaseSync(':memory:');
     const setup = db.exec(
       'CREATE TABLE data(key INTEGER PRIMARY KEY, val INTEGER) STRICT;'
     );
@@ -432,8 +423,7 @@ suite('StatementSync.prototype.run()', () => {
   });
 
   test('SQLite defaults to NULL for unbound parameters', (t) => {
-    const db = new DatabaseSync(':memory:');
-    t.after(() => { db.close(); });
+    using db = new DatabaseSync(':memory:');
     const setup = db.exec(
       'CREATE TABLE data(key INTEGER PRIMARY KEY, val INTEGER NOT NULL) STRICT;'
     );
@@ -469,8 +459,7 @@ suite('StatementSync.prototype.run()', () => {
   });
 
   test('SQLite defaults unbound ?NNN parameters', (t) => {
-    const db = new DatabaseSync(':memory:');
-    t.after(() => { db.close(); });
+    using db = new DatabaseSync(':memory:');
     const setup = db.exec(
       'CREATE TABLE data(key INTEGER PRIMARY KEY, val INTEGER NOT NULL) STRICT;'
     );
@@ -488,8 +477,7 @@ suite('StatementSync.prototype.run()', () => {
   });
 
   test('binds ?NNN params by position', (t) => {
-    const db = new DatabaseSync(':memory:');
-    t.after(() => { db.close(); });
+    using db = new DatabaseSync(':memory:');
     const setup = db.exec(
       'CREATE TABLE data(key INTEGER PRIMARY KEY, val INTEGER NOT NULL) STRICT;'
     );
@@ -513,8 +501,7 @@ suite('StatementSync.prototype.run()', () => {
 
 suite('StatementSync.prototype.sourceSQL', () => {
   test('equals input SQL', (t) => {
-    const db = new DatabaseSync(':memory:');
-    t.after(() => { db.close(); });
+    using db = new DatabaseSync(':memory:');
     const setup = db.exec(
       'CREATE TABLE types(key INTEGER PRIMARY KEY, val INTEGER) STRICT;'
     );
@@ -537,8 +524,7 @@ suite('StatementSync.prototype.sourceSQL', () => {
 
 suite('StatementSync.prototype.expandedSQL', () => {
   test('equals expanded SQL', (t) => {
-    const db = new DatabaseSync(':memory:');
-    t.after(() => { db.close(); });
+    using db = new DatabaseSync(':memory:');
     const setup = db.exec(
       'CREATE TABLE types(key INTEGER PRIMARY KEY, val INTEGER) STRICT;'
     );
@@ -772,8 +758,7 @@ suite('StatementSync.prototype.resetStats()', () => {
 
 suite('StatementSync.prototype.setReadBigInts()', () => {
   test('BigInts support can be toggled', (t) => {
-    const db = new DatabaseSync(':memory:');
-    t.after(() => { db.close(); });
+    using db = new DatabaseSync(':memory:');
     const setup = db.exec(`
       CREATE TABLE data(key INTEGER PRIMARY KEY, val INTEGER) STRICT;
       INSERT INTO data (key, val) VALUES (1, 42);
@@ -805,8 +790,7 @@ suite('StatementSync.prototype.setReadBigInts()', () => {
   });
 
   test('throws when input is not a boolean', (t) => {
-    const db = new DatabaseSync(':memory:');
-    t.after(() => { db.close(); });
+    using db = new DatabaseSync(':memory:');
     const setup = db.exec(
       'CREATE TABLE types(key INTEGER PRIMARY KEY, val INTEGER) STRICT;'
     );
@@ -821,8 +805,7 @@ suite('StatementSync.prototype.setReadBigInts()', () => {
   });
 
   test('BigInt is required for reading large integers', (t) => {
-    const db = new DatabaseSync(':memory:');
-    t.after(() => { db.close(); });
+    using db = new DatabaseSync(':memory:');
     const bad = db.prepare(`SELECT ${Number.MAX_SAFE_INTEGER} + 1`);
     t.assert.throws(() => {
       bad.get();
@@ -853,8 +836,7 @@ suite('StatementSync.prototype.setReadBigInts()', () => {
 
 suite('StatementSync.prototype.setReturnArrays()', () => {
   test('throws when input is not a boolean', (t) => {
-    const db = new DatabaseSync(':memory:');
-    t.after(() => { db.close(); });
+    using db = new DatabaseSync(':memory:');
     const setup = db.exec(
       'CREATE TABLE data(key INTEGER PRIMARY KEY, val INTEGER) STRICT;'
     );
@@ -883,8 +865,7 @@ suite('StatementSync.prototype.setReturnArrays()', () => {
 
 suite('StatementSync.prototype.get() with array output', () => {
   test('returns array row when setReturnArrays is true', (t) => {
-    const db = new DatabaseSync(':memory:');
-    t.after(() => { db.close(); });
+    using db = new DatabaseSync(':memory:');
     const setup = db.exec(`
       CREATE TABLE data(key INTEGER PRIMARY KEY, val TEXT) STRICT;
       INSERT INTO data (key, val) VALUES (1, 'one');
@@ -903,8 +884,7 @@ suite('StatementSync.prototype.get() with array output', () => {
 
   test('returns array rows with BigInts when both flags are set', (t) => {
     const expected = [1n, 9007199254740992n];
-    const db = new DatabaseSync(':memory:');
-    t.after(() => { db.close(); });
+    using db = new DatabaseSync(':memory:');
     const setup = db.exec(`
       CREATE TABLE big_data(id INTEGER, big_num INTEGER);
       INSERT INTO big_data VALUES (1, 9007199254740992);
@@ -922,8 +902,7 @@ suite('StatementSync.prototype.get() with array output', () => {
 
 suite('StatementSync.prototype.all() with array output', () => {
   test('returns array rows when setReturnArrays is true', (t) => {
-    const db = new DatabaseSync(':memory:');
-    t.after(() => { db.close(); });
+    using db = new DatabaseSync(':memory:');
     const setup = db.exec(`
       CREATE TABLE data(key INTEGER PRIMARY KEY, val TEXT) STRICT;
       INSERT INTO data (key, val) VALUES (1, 'one');
@@ -963,8 +942,7 @@ suite('StatementSync.prototype.all() with array output', () => {
       9,
       'text3',
     ];
-    const db = new DatabaseSync(':memory:');
-    t.after(() => { db.close(); });
+    using db = new DatabaseSync(':memory:');
     const setup = db.exec(`
       CREATE TABLE wide_table(
         col1 INTEGER, col2 TEXT, col3 REAL, col4 BLOB, col5 INTEGER,
@@ -988,8 +966,7 @@ suite('StatementSync.prototype.all() with array output', () => {
 
 suite('StatementSync.prototype.iterate() with array output', () => {
   test('iterates array rows when setReturnArrays is true', (t) => {
-    const db = new DatabaseSync(':memory:');
-    t.after(() => { db.close(); });
+    using db = new DatabaseSync(':memory:');
     const setup = db.exec(`
       CREATE TABLE data(key INTEGER PRIMARY KEY, val TEXT) STRICT;
       INSERT INTO data (key, val) VALUES (1, 'one');
@@ -1057,8 +1034,7 @@ suite('StatementSync.prototype.iterate() with array output', () => {
 
 suite('StatementSync.prototype.setAllowBareNamedParameters()', () => {
   test('bare named parameter support can be toggled', (t) => {
-    const db = new DatabaseSync(':memory:');
-    t.after(() => { db.close(); });
+    using db = new DatabaseSync(':memory:');
     const setup = db.exec(
       'CREATE TABLE data(key INTEGER PRIMARY KEY, val INTEGER) STRICT;'
     );
@@ -1083,8 +1059,7 @@ suite('StatementSync.prototype.setAllowBareNamedParameters()', () => {
   });
 
   test('throws when input is not a boolean', (t) => {
-    const db = new DatabaseSync(':memory:');
-    t.after(() => { db.close(); });
+    using db = new DatabaseSync(':memory:');
     const setup = db.exec(
       'CREATE TABLE data(key INTEGER PRIMARY KEY, val INTEGER) STRICT;'
     );
@@ -1113,8 +1088,7 @@ suite('StatementSync.prototype.setAllowBareNamedParameters()', () => {
 
 suite('options.readBigInts', () => {
   test('BigInts are returned when input is true', (t) => {
-    const db = new DatabaseSync(':memory:');
-    t.after(() => { db.close(); });
+    using db = new DatabaseSync(':memory:');
     const setup = db.exec(`
       CREATE TABLE data(key INTEGER PRIMARY KEY, val INTEGER) STRICT;
       INSERT INTO data (key, val) VALUES (1, 42);
@@ -1126,8 +1100,7 @@ suite('options.readBigInts', () => {
   });
 
   test('numbers are returned when input is false', (t) => {
-    const db = new DatabaseSync(':memory:');
-    t.after(() => { db.close(); });
+    using db = new DatabaseSync(':memory:');
     const setup = db.exec(`
       CREATE TABLE data(key INTEGER PRIMARY KEY, val INTEGER) STRICT;
       INSERT INTO data (key, val) VALUES (1, 42);
@@ -1139,8 +1112,7 @@ suite('options.readBigInts', () => {
   });
 
   test('throws when input is not a boolean', (t) => {
-    const db = new DatabaseSync(':memory:');
-    t.after(() => { db.close(); });
+    using db = new DatabaseSync(':memory:');
     const setup = db.exec(
       'CREATE TABLE data(key INTEGER PRIMARY KEY, val INTEGER) STRICT;'
     );
@@ -1154,8 +1126,7 @@ suite('options.readBigInts', () => {
   });
 
   test('setReadBigInts can override prepare option', (t) => {
-    const db = new DatabaseSync(':memory:');
-    t.after(() => { db.close(); });
+    using db = new DatabaseSync(':memory:');
     const setup = db.exec(`
       CREATE TABLE data(key INTEGER PRIMARY KEY, val INTEGER) STRICT;
       INSERT INTO data (key, val) VALUES (1, 42);
@@ -1171,8 +1142,7 @@ suite('options.readBigInts', () => {
 
 suite('options.returnArrays', () => {
   test('arrays are returned when input is true', (t) => {
-    const db = new DatabaseSync(':memory:');
-    t.after(() => { db.close(); });
+    using db = new DatabaseSync(':memory:');
     const setup = db.exec(`
       CREATE TABLE data(key INTEGER PRIMARY KEY, val TEXT) STRICT;
       INSERT INTO data (key, val) VALUES (1, 'one');
@@ -1187,8 +1157,7 @@ suite('options.returnArrays', () => {
   });
 
   test('objects are returned when input is false', (t) => {
-    const db = new DatabaseSync(':memory:');
-    t.after(() => { db.close(); });
+    using db = new DatabaseSync(':memory:');
     const setup = db.exec(`
       CREATE TABLE data(key INTEGER PRIMARY KEY, val TEXT) STRICT;
       INSERT INTO data (key, val) VALUES (1, 'one');
@@ -1203,8 +1172,7 @@ suite('options.returnArrays', () => {
   });
 
   test('throws when input is not a boolean', (t) => {
-    const db = new DatabaseSync(':memory:');
-    t.after(() => { db.close(); });
+    using db = new DatabaseSync(':memory:');
     const setup = db.exec(
       'CREATE TABLE data(key INTEGER PRIMARY KEY, val TEXT) STRICT;'
     );
@@ -1218,8 +1186,7 @@ suite('options.returnArrays', () => {
   });
 
   test('setReturnArrays can override prepare option', (t) => {
-    const db = new DatabaseSync(':memory:');
-    t.after(() => { db.close(); });
+    using db = new DatabaseSync(':memory:');
     const setup = db.exec(`
       CREATE TABLE data(key INTEGER PRIMARY KEY, val TEXT) STRICT;
       INSERT INTO data (key, val) VALUES (1, 'one');
@@ -1236,8 +1203,7 @@ suite('options.returnArrays', () => {
   });
 
   test('all() returns arrays when input is true', (t) => {
-    const db = new DatabaseSync(':memory:');
-    t.after(() => { db.close(); });
+    using db = new DatabaseSync(':memory:');
     const setup = db.exec(`
       CREATE TABLE data(key INTEGER PRIMARY KEY, val TEXT) STRICT;
       INSERT INTO data (key, val) VALUES (1, 'one');
@@ -1256,8 +1222,7 @@ suite('options.returnArrays', () => {
   });
 
   test('iterate() returns arrays when input is true', (t) => {
-    const db = new DatabaseSync(':memory:');
-    t.after(() => { db.close(); });
+    using db = new DatabaseSync(':memory:');
     const setup = db.exec(`
       CREATE TABLE data(key INTEGER PRIMARY KEY, val TEXT) STRICT;
       INSERT INTO data (key, val) VALUES (1, 'one');
@@ -1278,8 +1243,7 @@ suite('options.returnArrays', () => {
 
 suite('options.allowBareNamedParameters', () => {
   test('bare named parameters are allowed when input is true', (t) => {
-    const db = new DatabaseSync(':memory:');
-    t.after(() => { db.close(); });
+    using db = new DatabaseSync(':memory:');
     const setup = db.exec(
       'CREATE TABLE data(key INTEGER PRIMARY KEY, val INTEGER) STRICT;'
     );
@@ -1295,8 +1259,7 @@ suite('options.allowBareNamedParameters', () => {
   });
 
   test('bare named parameters throw when input is false', (t) => {
-    const db = new DatabaseSync(':memory:');
-    t.after(() => { db.close(); });
+    using db = new DatabaseSync(':memory:');
     const setup = db.exec(
       'CREATE TABLE data(key INTEGER PRIMARY KEY, val INTEGER) STRICT;'
     );
@@ -1314,8 +1277,7 @@ suite('options.allowBareNamedParameters', () => {
   });
 
   test('throws when input is not a boolean', (t) => {
-    const db = new DatabaseSync(':memory:');
-    t.after(() => { db.close(); });
+    using db = new DatabaseSync(':memory:');
     const setup = db.exec(
       'CREATE TABLE data(key INTEGER PRIMARY KEY, val INTEGER) STRICT;'
     );
@@ -1332,8 +1294,7 @@ suite('options.allowBareNamedParameters', () => {
   });
 
   test('setAllowBareNamedParameters can override prepare option', (t) => {
-    const db = new DatabaseSync(':memory:');
-    t.after(() => { db.close(); });
+    using db = new DatabaseSync(':memory:');
     const setup = db.exec(
       'CREATE TABLE data(key INTEGER PRIMARY KEY, val INTEGER) STRICT;'
     );
@@ -1423,3 +1384,42 @@ suite('StatementSync.prototype[Symbol.dispose]()', () => {
     db.close();
   });
 });
+
+suite('options.persistent', () => {
+  test('statement executes correctly when persistent is true', (t) => {
+    using db = new DatabaseSync(':memory:');
+    db.exec('CREATE TABLE data(key INTEGER PRIMARY KEY, val INTEGER) STRICT;');
+    db.exec('INSERT INTO data (key, val) VALUES (1, 42);');
+    using stmt = db.prepare('SELECT val FROM data', { persistent: true });
+    t.assert.deepStrictEqual(stmt.get(), { __proto__: null, val: 42 });
+  });
+
+  test('statement executes correctly when persistent is false', (t) => {
+    using db = new DatabaseSync(':memory:');
+    db.exec('CREATE TABLE data(key INTEGER PRIMARY KEY, val INTEGER) STRICT;');
+    db.exec('INSERT INTO data (key, val) VALUES (1, 42);');
+    using stmt = db.prepare('SELECT val FROM data', { persistent: false });
+    t.assert.deepStrictEqual(stmt.get(), { __proto__: null, val: 42 });
+  });
+
+  test('throws when input is not a boolean', (t) => {
+    using db = new DatabaseSync(':memory:');
+    t.assert.throws(() => {
+      db.prepare('SELECT 1', { persistent: 'yes' });
+    }, {
+      code: 'ERR_INVALID_ARG_TYPE',
+      message: /The "options\.persistent" argument must be a boolean/,
+    });
+  });
+
+  test('can be combined with other options', (t) => {
+    using db = new DatabaseSync(':memory:');
+    db.exec('CREATE TABLE data(key INTEGER PRIMARY KEY, val INTEGER) STRICT;');
+    db.exec('INSERT INTO data (key, val) VALUES (1, 42);');
+    using stmt = db.prepare(
+      'SELECT val FROM data',
+      { persistent: true, readBigInts: true }
+    );
+    t.assert.deepStrictEqual(stmt.get(), { __proto__: null, val: 42n });
+  });
+});
diff --git a/test/parallel/test-sqlite-udf-close.js b/test/parallel/test-sqlite-udf-close.js
index 86794029b457..cb11e50a7f7a 100644
--- a/test/parallel/test-sqlite-udf-close.js
+++ b/test/parallel/test-sqlite-udf-close.js
@@ -36,4 +36,207 @@ for (const method of ['all', 'get', 'run', 'iterate']) {
     assert.strictEqual(db.isOpen, true);
     db.close();
   });
+
+  // Finalizing the statement being stepped frees the virtual machine that
+  // sqlite3_step() is still running, so this must throw rather than crash.
+  test(`statement.close() from a UDF during statement.${method}()`, () => {
+    const db = new DatabaseSync(':memory:');
+    db.exec(`
+      CREATE TABLE data (value INTEGER);
+      INSERT INTO data VALUES (1), (2), (3);
+    `);
+
+    let statement;
+    db.function('close_stmt', (value) => {
+      statement.close();
+      return value;
+    });
+
+    statement = db.prepare('SELECT close_stmt(value) FROM data');
+    assert.throws(() => {
+      if (method === 'iterate') {
+        for (const row of statement.iterate()) {
+          assert.ok(row);
+        }
+      } else {
+        statement[method]();
+      }
+    }, {
+      code: 'ERR_INVALID_STATE',
+      message: 'statement is already being executed',
+    });
+
+    db.close();
+  });
+
+  // Re-running the statement being stepped resets its virtual machine
+  // mid-execution, which is the same use-after-free as finalizing it.
+  for (const reentrant of ['run', 'get', 'all', 'iterate']) {
+    test(`statement.${reentrant}() from a UDF during ` +
+         `statement.${method}()`, () => {
+      const db = new DatabaseSync(':memory:');
+      db.exec(`
+        CREATE TABLE data (value INTEGER, padding TEXT);
+        INSERT INTO data VALUES (1, '${'x'.repeat(400)}'),
+                                (2, '${'y'.repeat(400)}'),
+                                (3, '${'z'.repeat(400)}');
+      `);
+
+      let statement;
+      let thrown;
+      db.function('reenter', (value) => {
+        if (thrown === undefined) {
+          try {
+            statement[reentrant]();
+            thrown = null;
+          } catch (err) {
+            thrown = err;
+          }
+        }
+        return value;
+      });
+
+      statement = db.prepare('SELECT reenter(value), padding FROM data');
+      if (method === 'iterate') {
+        for (const row of statement.iterate()) {
+          assert.ok(row);
+        }
+      } else {
+        statement[method]();
+      }
+
+      assert.ok(thrown, `${reentrant}() was not rejected`);
+      assert.strictEqual(thrown.code, 'ERR_INVALID_STATE');
+      assert.strictEqual(thrown.message, 'statement is already being executed');
+
+      db.close();
+    });
+  }
+
+  // Tag store methods resolve to a cached statement, which may be the one
+  // currently being stepped. Each reentrant method has its own guard, so all
+  // four are exercised.
+  for (const reentrant of ['run', 'get', 'all', 'iterate']) {
+    test(`tag store ${reentrant} reentry during statement.${method}()`, () => {
+      const db = new DatabaseSync(':memory:');
+      const sql = db.createTagStore(10);
+      db.exec(`
+        CREATE TABLE data (value INTEGER, padding TEXT);
+        INSERT INTO data VALUES (1, '${'x'.repeat(400)}'),
+                                (2, '${'y'.repeat(400)}');
+      `);
+
+      let thrown;
+      db.function('reenter_tag', (value) => {
+        if (thrown === undefined) {
+          try {
+            // The identical tagged literal resolves to the same cached
+            // statement that is mid-execution.
+            // All four reject at call time, iterate() included, so the
+            // result is never consumed.
+            // eslint-disable-next-line no-unused-expressions
+            sql[reentrant]`SELECT reenter_tag(value), padding FROM data`;
+            thrown = null;
+          } catch (err) {
+            thrown = err;
+          }
+        }
+        return value;
+      });
+
+      if (method === 'iterate') {
+        for (const row of sql.iterate`SELECT reenter_tag(value), padding FROM data`) {
+          assert.ok(row);
+        }
+      } else {
+        // eslint-disable-next-line no-unused-expressions
+        sql[method]`SELECT reenter_tag(value), padding FROM data`;
+      }
+
+      assert.ok(thrown, `tag store ${reentrant} reentry was not rejected`);
+      assert.strictEqual(thrown.code, 'ERR_INVALID_STATE');
+      assert.strictEqual(thrown.message,
+                         'statement is already being executed');
+
+      db.close();
+    });
+  }
+
+  // A UDF may prepare and finalize its own helper statements. Only the
+  // statement being stepped is off limits.
+  test(`UDF finalizes its own statement during statement.${method}()`, () => {
+    const db = new DatabaseSync(':memory:');
+    db.exec(`
+      CREATE TABLE data (value INTEGER);
+      INSERT INTO data VALUES (1), (2), (3);
+      CREATE TABLE lookup (key INTEGER, label TEXT);
+      INSERT INTO lookup VALUES (1, 'one'), (2, 'two'), (3, 'three');
+    `);
+
+    db.function('lookup_label', (value) => {
+      const helper = db.prepare('SELECT label FROM lookup WHERE key = ?');
+      const label = helper.get(value).label;
+      helper.close();
+      return label;
+    });
+
+    const statement = db.prepare('SELECT lookup_label(value) AS l FROM data');
+    if (method === 'iterate') {
+      const labels = [];
+      for (const row of statement.iterate()) {
+        labels.push(row.l);
+      }
+      assert.deepStrictEqual(labels, ['one', 'two', 'three']);
+    } else if (method === 'all') {
+      assert.deepStrictEqual(statement.all().map((r) => r.l),
+                             ['one', 'two', 'three']);
+    } else if (method === 'get') {
+      assert.strictEqual(statement.get().l, 'one');
+    } else {
+      statement.run();
+    }
+
+    db.close();
+  });
+}
+
+// iterator.return() resets the statement it is iterating, and next() steps it
+// again, so both reach the virtual machine that is mid-execution.
+for (const op of ['next', 'return']) {
+  test(`iterator.${op}() from a UDF during iteration`, () => {
+    const db = new DatabaseSync(':memory:');
+    db.exec(`
+      CREATE TABLE data (value INTEGER, padding TEXT);
+      INSERT INTO data VALUES (1, '${'x'.repeat(400)}'),
+                              (2, '${'y'.repeat(400)}'),
+                              (3, '${'z'.repeat(400)}');
+    `);
+
+    let iterator;
+    let thrown;
+    db.function('reenter_iter', (value) => {
+      if (thrown === undefined && iterator !== undefined) {
+        try {
+          iterator[op]();
+          thrown = null;
+        } catch (err) {
+          thrown = err;
+        }
+      }
+      return value;
+    });
+
+    const statement = db.prepare(
+      'SELECT reenter_iter(value) AS v, padding FROM data');
+    iterator = statement.iterate();
+    for (const row of iterator) {
+      assert.ok(row);
+    }
+
+    assert.ok(thrown, `iterator.${op}() was not rejected`);
+    assert.strictEqual(thrown.code, 'ERR_INVALID_STATE');
+    assert.strictEqual(thrown.message, 'statement is already being executed');
+
+    db.close();
+  });
 }
diff --git a/test/parallel/test-stream-flatMap.js b/test/parallel/test-stream-flatMap.js
index 0e55119f7a76..a75672e73c8d 100644
--- a/test/parallel/test-stream-flatMap.js
+++ b/test/parallel/test-stream-flatMap.js
@@ -72,10 +72,23 @@ function oneTo5() {
 
 {
   // Concurrency + AbortSignal
+  // Two mappers are started concurrently and block until their signal
+  // is aborted. Aborting while both are in flight must cancel them and
+  // reject the iteration, without ever starting a third mapper.
   const ac = new AbortController();
-  const stream = oneTo5().flatMap(common.mustNotCall(async (_, { signal }) => {
-    await setTimeout(100, { signal });
-  }), { signal: ac.signal, concurrency: 2 });
+  const stream = oneTo5().flatMap(common.mustCall(async (x, { signal }) => {
+    if (x === 2) {
+      // Both mappers allowed by `concurrency` are now in flight.
+      ac.abort();
+    }
+    const { promise, reject } = Promise.withResolvers();
+    if (signal.aborted) {
+      reject(signal.reason);
+    }
+    signal.addEventListener('abort', () => reject(signal.reason), { once: true });
+    // Promise is expected to reject.
+    await promise;
+  }, 2), { signal: ac.signal, concurrency: 2 });
   // pump
   assert.rejects(async () => {
     for await (const item of stream) {
@@ -85,10 +98,6 @@ function oneTo5() {
   }, {
     name: 'AbortError',
   }).then(common.mustCall());
-
-  queueMicrotask(() => {
-    ac.abort();
-  });
 }
 
 {
diff --git a/test/parallel/test-stream-readable-async-iterators.js b/test/parallel/test-stream-readable-async-iterators.js
index 8081b35191a1..f9bcaea6057b 100644
--- a/test/parallel/test-stream-readable-async-iterators.js
+++ b/test/parallel/test-stream-readable-async-iterators.js
@@ -869,5 +869,125 @@ async function tests() {
   }));
 }
 
+{
+  // Thenable chunks are awaited before delivery.
+  (async () => {
+    const r = new Readable({ objectMode: true, read() {} });
+    r.push(Promise.resolve('unwrapped'));
+    r.push(null);
+
+    const it = r[Symbol.asyncIterator]();
+    const { value, done } = await it.next();
+    assert.strictEqual(done, false);
+    assert.strictEqual(value, 'unwrapped');
+  })().then(common.mustCall());
+}
+
+{
+  // A rejected thenable chunk tears down the iterator and the stream.
+  (async () => {
+    const r = new Readable({ objectMode: true, read() {} });
+    const rejected = Promise.reject(new Error('kaboom'));
+    rejected.catch(() => {});
+    r.push(rejected);
+    r.push(null);
+
+    const it = r[Symbol.asyncIterator]();
+    await assert.rejects(it.next(), { message: 'kaboom' });
+    assert.strictEqual((await it.next()).done, true);
+    assert.strictEqual(r.destroyed, true);
+  })().then(common.mustCall());
+}
+
+{
+  // throw() rejects with the passed error, destroys the stream and
+  // completes the iterator.
+  (async () => {
+    const r = new Readable({ objectMode: true, read() {} });
+    r.push('a');
+
+    const it = r[Symbol.asyncIterator]();
+    assert.strictEqual((await it.next()).value, 'a');
+    await assert.rejects(it.throw(new Error('kaboom')), { message: 'kaboom' });
+    assert.strictEqual(r.destroyed, true);
+    assert.strictEqual((await it.next()).done, true);
+  })().then(common.mustCall());
+}
+
+{
+  // throw() before the first next() completes the iterator without
+  // touching the stream.
+  (async () => {
+    const r = new Readable({ objectMode: true, read() {} });
+    const it = r[Symbol.asyncIterator]();
+    await assert.rejects(it.throw(new Error('kaboom')), { message: 'kaboom' });
+    assert.strictEqual(r.destroyed, false);
+    assert.strictEqual(r.listenerCount('readable'), 0);
+    assert.strictEqual((await it.next()).done, true);
+    r.destroy();
+  })().then(common.mustCall());
+}
+
+{
+  // Concurrent next() calls while waiting for data are served in order.
+  (async () => {
+    const r = new Readable({ objectMode: true, read() {} });
+    const it = r[Symbol.asyncIterator]();
+
+    const p1 = it.next();
+    const p2 = it.next();
+    r.push('a');
+    r.push('b');
+    r.push(null);
+
+    assert.strictEqual((await p1).value, 'a');
+    assert.strictEqual((await p2).value, 'b');
+    assert.strictEqual((await it.next()).done, true);
+  })().then(common.mustCall());
+}
+
+{
+  // return() while a next() is pending is processed after the pending
+  // next() settles, and destroys the stream.
+  (async () => {
+    const r = new Readable({ objectMode: true, read() {} });
+    const it = r[Symbol.asyncIterator]();
+
+    const p1 = it.next();
+    const p2 = it.return();
+    r.push('a');
+
+    assert.strictEqual((await p1).value, 'a');
+    assert.deepStrictEqual(await p2, { done: true, value: undefined });
+    assert.strictEqual(r.destroyed, true);
+  })().then(common.mustCall());
+}
+
+{
+  // Draining a large number of queued next() calls must not overflow
+  // the call stack.
+  // Refs: https://github.com/nodejs/node/pull/64447#discussion_r3566240419
+  (async () => {
+    const count = 20_000;
+    const r = new Readable({ objectMode: true, read() {} });
+    const it = r[Symbol.asyncIterator]();
+
+    const requests = [];
+    for (let i = 0; i < count; i++) {
+      requests.push(it.next());
+    }
+    for (let i = 0; i < count; i++) {
+      r.push(i);
+    }
+    r.push(null);
+
+    const results = await Promise.all(requests);
+    for (let i = 0; i < count; i++) {
+      assert.deepStrictEqual(results[i], { done: false, value: i });
+    }
+    assert.strictEqual((await it.next()).done, true);
+  })().then(common.mustCall());
+}
+
 // To avoid missing some tests if a promise does not resolve
 tests().then(common.mustCall());