chore: fix some potential race conditions

Author: asbisen

src/context.jl

@@ -122,16 +122,23 @@ from the pool), ensure all registered sources are applied to it, call `f`,
 and release the connection back to the pool if needed.
 """
 function _with_conn(f::Function, ctx::QueryContext)
-    ctx._closed && throw(QueryError("QueryContext has been closed."))
-    if ctx._pool !== nothing
+    # Read _closed, _pool, and _conn atomically under the same lock that
+    # close!(ctx) holds when it nulls them out.  This eliminates the TOCTOU
+    # window where _closed passes but _conn is set to nothing before f(conn).
+    pool, conn = lock(ctx._lock) do
+        ctx._closed && throw(QueryError("QueryContext has been closed."))
+        (ctx._pool, ctx._conn)
+    end
+    if pool !== nothing
         # Pool path: acquire! already calls _ensure_sources_applied!, which
         # reads pool.sources — a mirror of ctx.sources kept in sync by
         # register!/deregister!.  No secondary source check is needed here.
-        with_connection(ctx._pool) do conn
-            f(conn)
+        with_connection(pool) do c
+            f(c)
         end
     else
-        # Single connection path
-        f(ctx._conn)
+        # Single connection path: conn is a local reference captured before
+        # the lock was released, so a concurrent close! cannot null it out.
+        f(conn)
     end
 end

src/query.jl

@@ -132,7 +132,7 @@ function query(ctx::QueryContext, sql::String, args...; kwargs...)::QueryResult
     processed_sql, params = normalise_params(sql, args, kwargs)
     _with_conn(ctx) do conn
         df, elapsed = _run(conn, ctx.config, processed_sql, params)
-        QueryResult(df, elapsed, sql)
+        QueryResult(df, elapsed, processed_sql)
     end
 end
 
@@ -241,6 +241,7 @@ end
 ```
 """
 function stream(ctx::QueryContext, sql::String, args...; batch_size::Int=10_000, kwargs...)::Channel{DataFrame}
+    batch_size > 0 || throw(ArgumentError("batch_size must be a positive integer, got $batch_size"))
     processed_sql, params = normalise_params(sql, args, kwargs)
     Channel{DataFrame}(2) do ch
         _with_conn(ctx) do conn

test/runtests.jl

@@ -58,6 +58,9 @@ global_logger(ConsoleLogger(stderr, Logging.Warn))
         @test r.elapsed_ns > 0
         @test elapsed_ms(r) >= 0.0
         @test DataFrame(r) isa DataFrame
+        # Named params: stored SQL must use ? placeholders, not :name tokens
+        r2 = query(ctx, "SELECT :val AS n", val=42)
+        @test r2.sql == "SELECT ? AS n"
         close!(ctx)
     end
 
@@ -217,6 +220,13 @@ global_logger(ConsoleLogger(stderr, Logging.Warn))
     end
 
     # ── Streaming ──────────────────────────────────────────────────────────────
+    @testset "stream batch_size validation" begin
+        ctx = QueryContext()
+        @test_throws ArgumentError stream(ctx, "SELECT 1"; batch_size=0)
+        @test_throws ArgumentError stream(ctx, "SELECT 1"; batch_size=-1)
+        close!(ctx)
+    end
+
     @testset "stream in batches" begin
         ctx = QueryContext()
         execute!(ctx, "CREATE TABLE big AS SELECT generate_series AS n FROM generate_series(1, 100)")