ConnectionPool can return a busy reader and pin WAL while autocommit is true

[daniel-vacic]

Summary

ConnectionPool can return a reader to the pool while a statement on that
connection is still busy. The statement keeps a WAL read snapshot even though
SQLite reports autocommit, so later checkpoints cannot reclaim all frames.

This is reproducible on current main (654bac2) with powersync_core 0.5.1.
It is not sufficient to use sqlite3_get_autocommit() as a pool-cleanliness
test: a stepped SELECT can hold a read transaction while autocommit is true.

Minimal reproduction

The test uses only public powersync and rusqlite APIs plus SQLite's FFI to
deliberately leave one statement stepped:

use powersync::{ConnectionPool, env::PowerSyncEnvironment};
use rusqlite::ffi::{
    SQLITE_OK, SQLITE_ROW, sqlite3_finalize, sqlite3_prepare_v2, sqlite3_step,
    sqlite3_stmt, sqlite3_stmt_busy,
};
use std::{ffi::CString, ptr};

unsafe fn step_without_reset(
    connection: &rusqlite::Connection,
    sql: &str,
) -> *mut sqlite3_stmt {
    let sql = CString::new(sql).unwrap();
    let mut statement = ptr::null_mut();
    assert_eq!(unsafe {
        sqlite3_prepare_v2(
            connection.handle(),
            sql.as_ptr(),
            -1,
            &mut statement,
            ptr::null_mut(),
        )
    }, SQLITE_OK);
    assert_eq!(unsafe { sqlite3_step(statement) }, SQLITE_ROW);
    statement
}

#[test]
fn returned_reader_must_not_keep_a_busy_statement_or_pin_the_wal() {
    PowerSyncEnvironment::powersync_auto_extension().unwrap();
    let path = std::env::temp_dir().join("powersync-pool-wal-repro.db");
    let _ = std::fs::remove_file(&path);
    let writer = rusqlite::Connection::open(&path).unwrap();
    writer.execute_batch(
        "PRAGMA journal_mode=WAL;
         PRAGMA wal_autocheckpoint=0;
         CREATE TABLE items(id INTEGER PRIMARY KEY, value TEXT);
         INSERT INTO items(value) VALUES ('snapshot');",
    ).unwrap();
    let reader = rusqlite::Connection::open(&path).unwrap();
    let pool = ConnectionPool::wrap_connections(writer, [reader]);

    let reader = pool.reader_sync();
    let statement = unsafe { step_without_reset(&reader, "SELECT * FROM items") };
    assert!(reader.is_autocommit());
    drop(reader);

    let writer = pool.writer_sync();
    writer.execute_batch(
        "BEGIN IMMEDIATE;
         WITH RECURSIVE n(x) AS (
           VALUES(1) UNION ALL SELECT x + 1 FROM n WHERE x < 2000
         )
         INSERT INTO items(value)
         SELECT printf('value-%04d', x) FROM n;
         COMMIT;",
    ).unwrap();
    let (_, log, copied): (i64, i64, i64) = writer.query_row(
        "PRAGMA wal_checkpoint(PASSIVE)",
        [],
        |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
    ).unwrap();
    drop(writer);

    let returned_reader = pool.reader_sync();
    assert_ne!(unsafe { sqlite3_stmt_busy(statement) }, 0);
    unsafe { sqlite3_finalize(statement) };
    drop(returned_reader);

    assert_eq!(log, copied, "released lease stranded WAL frames");
}

Run with:

cargo test -p powersync --test wal_pool_repro -- --nocapture

Observed:

pool returned a reader with a stepped statement still busy:
log=14, checkpointed=3; after explicit finalize log=14, checkpointed=14

Why the current lifecycle permits it

OwnedConnectionLease::drop sends a reader straight back through
release_reader; it does not inspect/reset busy statements or otherwise
quarantine a dirty handle. The writer path similarly publishes notifications
and releases the mutex without a pool-boundary cleanliness check.

An explicit transaction guard does not cover this case because the read
transaction is implicit and sqlite3_get_autocommit() remains true.

Suggested direction

Two layers seem useful:

1. A scoped statement/transaction API that owns reset/finalize on every success,
   error, cancellation and panic path.
2. A fail-closed pool-boundary check using sqlite3_next_stmt() plus
   sqlite3_stmt_busy(), with reset/finalize (or quarantine) before the physical
   connection becomes available again.

I can submit a focused PR with the regression and scoped API if that direction
fits the project.