Preserve enclosing transactions with nested savepoints

docs/src/index.md

@@ -73,7 +73,19 @@ DBInterface.close!(stmt)
 ```
 
 `DBInterface.transaction(conn) do ... end` wraps the block in `START TRANSACTION`/`COMMIT`
-(`ROLLBACK` on an exception). `DBInterface.executemultiple` iterates the result sets of a
+(`ROLLBACK` on an exception). If a transaction is already active, including one started
+with SQL, the block uses a savepoint. Nested success leaves the outer transaction open;
+nested failure rolls back only that block's changes. This also lets `executemany` run
+inside a caller's transaction. The connection lock stays held throughout the block;
+do not wait on another task that needs the same connection.
+
+Statements that implicitly commit, such as most DDL, and explicit transaction-control
+statements inside the block are not protected. For example, `MySQL.load` creates a table
+before inserting rows; that creation can commit an existing transaction. Savepoint names
+beginning with `_mysqljl_` are reserved. A failed rollback closes the connection and reports
+both the original and cleanup errors in a `CompositeException`.
+
+`DBInterface.executemultiple` iterates the result sets of a
 `CALL` or of a multi-statement string (`multi_statements=true`) as distinct cursors.
 `MySQL.load(table, conn, name)` creates a table from a Tables.jl source's schema and
 inserts its rows in batches. Close the connection with `DBInterface.close!(conn)`.

src/connection.jl

@@ -264,7 +264,10 @@ end
 
 # Runs a statement that must answer with OK (no result set) and returns the OK packet.
 function execute_ok!(conn::Connection, sql::AbstractString)
-    s = begin_command!(conn)
+    return execute_ok!(begin_command!(conn), sql)
+end
+
+function execute_ok!(s::P.Session, sql::AbstractString)
     P.query!(s, sql)
     resp = P.read_command_response!(s)
     if !(resp isa P.OKPacket)
@@ -318,30 +321,55 @@ server_kind(conn::Connection) = return (session(conn).server::P.ServerInfo).kind
     DBInterface.transaction(f, conn)
 
 Runs `f()` inside `START TRANSACTION` / `COMMIT` (or `ROLLBACK` when `f` throws) and returns
-`f()`'s value. The connection lock is held for the whole callback: other tasks block until
-the transaction ends, so `f` must not wait on tasks that need this connection.
+`f()`'s value. Inside an existing transaction, including one started with SQL, uses a
+savepoint instead: success releases it, and failure rolls back only this block's changes.
+Savepoint names beginning with `_mysqljl_` are reserved for the driver.
+
+The connection lock is held for the whole callback: other tasks block until the block
+ends, so `f` must not wait on tasks that need this connection. Statements that implicitly
+commit (such as most DDL) and explicit transaction-control statements inside `f` are not
+protected by the block. If rollback fails, closes the connection and throws a
+`CompositeException` containing the original error, the cleanup error, and any close error.
 """
 function DBInterface.transaction(f, conn::Connection)
     lock(conn.lock)
-    owns_transaction = false
+    previous_owner = conn.transaction_owner
     try
-        conn.transaction_owner === nothing || throw(MySQLInterfaceError("a transaction is already active on this connection"))
-        execute_ok!(conn, "START TRANSACTION")
+        # Draining a previous result can update the server's transaction status.
+        s = begin_command!(conn)
+        if previous_owner !== nothing && !P.in_transaction(s.status)
+            throw(MySQLInterfaceError("the enclosing transaction has already ended"))
+        end
+        savepoint = if previous_owner !== nothing || P.in_transaction(s.status)
+            conn.next_token += 1
+            string("_mysqljl_", conn.next_token)
+        else
+            nothing
+        end
+        execute_ok!(s, savepoint === nothing ? "START TRANSACTION" : "SAVEPOINT $savepoint")
         conn.transaction_owner = current_task()
-        owns_transaction = true
         try
             result = f()
-            execute_ok!(conn, "COMMIT")
+            execute_ok!(conn, savepoint === nothing ? "COMMIT" : "RELEASE SAVEPOINT $savepoint")
             return result
-        catch
+        catch original_error
             try
-                execute_ok!(conn, "ROLLBACK")
-            catch
+                execute_ok!(conn, savepoint === nothing ? "ROLLBACK" : "ROLLBACK TO SAVEPOINT $savepoint")
+                savepoint === nothing || execute_ok!(conn, "RELEASE SAVEPOINT $savepoint")
+            catch cleanup_error
+                # An unconfirmed rollback must not leave a session available for reuse.
+                errors = Any[original_error, cleanup_error]
+                try
+                    DBInterface.close!(conn)
+                catch close_error
+                    push!(errors, close_error)
+                end
+                throw(CompositeException(errors))
             end
             rethrow()
         end
     finally
-        owns_transaction && (conn.transaction_owner = nothing)
+        conn.transaction_owner = previous_owner
         unlock(conn.lock)
     end
 end

test/mysql_trim_workload.jl

@@ -1,6 +1,7 @@
 # JuliaC --trim=safe workload: drives the main MySQL.jl entrypoints — connect (handshake,
 # auth, charset bootstrap), text-protocol execute (buffered and streaming), prepared
-# statements (binary protocol), one-shot parameterized execute, ping, escape, and close —
+# statements (binary protocol), one-shot parameterized execute, transaction scopes,
+# rollback failure, ping, escape, and close —
 # against an in-process scripted server on a loopback Reseau TCP listener, so the
 # executable needs no database and runs on every platform.
 #
@@ -162,6 +163,8 @@ function serve_connection!(conn)::Nothing
     send_packet(conn, 0, greeting_payload())
     seq, _ = read_packet(conn)                 # handshake response (auth ignored)
     send_packet(conn, seq + 1, ok_payload())
+    status = P.SERVER_STATUS_AUTOCOMMIT
+    fail_rollback = false
     while true
         local cmd
         local payload
@@ -182,9 +185,16 @@ function serve_connection!(conn)::Nothing
             sql = String(payload[2:end])
             if occursin("FROM people", sql)
                 send_resultset(conn, people_cols(), people_rows())
+            elseif fail_rollback && startswith(sql, "ROLLBACK")
+                err = UInt8[0xff]
+                P.write_u16!(err, 1305)
+                append!(err, codeunits("#HY000scripted rollback failure"))
+                send_packet(conn, 1, err)
             else
-                # SET NAMES, CREATE TABLE, START TRANSACTION, COMMIT, INSERT, ...
-                send_packet(conn, 1, ok_payload(; affected=occursin("INSERT", sql) ? 1 : 0))
+                sql == "START TRANSACTION" && (status |= P.SERVER_STATUS_IN_TRANS)
+                sql in ("COMMIT", "ROLLBACK") && (status = P.SERVER_STATUS_AUTOCOMMIT)
+                sql == "FAIL ROLLBACK" && (fail_rollback = true)
+                send_packet(conn, 1, ok_payload(; affected=occursin("INSERT", sql) ? 1 : 0, status))
             end
         elseif cmd == P.COM_STMT_PREPARE
             sql = String(payload[2:end])
@@ -216,10 +226,9 @@ function check(cond::Bool, what::String)::Nothing
     return nothing
 end
 
-function run_workload(port::Int)::Nothing
-    # no connect_timeout: Reseau <= 1.4.1's deadline-armed dial parks on tasks a trimmed
-    # build never ran (fixed in Reseau#151); the scripted loopback peer answers immediately
-    # anyway — re-add once the fixed Reseau is released
+function run_workload(port::Int, value::Int)::Int
+    # The scripted loopback peer answers immediately; this workload does not
+    # exercise connection deadlines.
     conn = DBInterface.connect(MySQL.Connection, "127.0.0.1", "root", "secret"; port=port, ssl_mode=:disabled)
     try
         check(isopen(conn)::Bool, "connection is open")
@@ -275,11 +284,45 @@ function run_workload(port::Int)::Nothing
         check(MySQL.ping(conn), "ping")
         check(MySQL.escape(conn, "a'b") == "a\\'b", "escape")
         check(MySQL.escape_identifier("weird`name") == "`weird``name`", "escape_identifier")
+        return run_transaction_workload(conn, value)
     finally
         DBInterface.close!(conn)
     end
-    check(!(isopen(conn)::Bool), "connection closed")
-    return nothing
+end
+
+function run_transaction_workload(conn::MySQL.Connection, value::Int)::Int
+    result = DBInterface.transaction(conn) do
+        inner = DBInterface.transaction(() -> value, conn)
+        inner * 2 + 1
+    end
+    DBInterface.execute(conn, "START TRANSACTION")
+    marker = ErrorException("scripted callback failure")
+    caught = try
+        DBInterface.transaction(() -> throw(marker), conn)
+        false
+    catch err
+        err === marker
+    end
+    check(caught, "original callback error")
+    check(P.in_transaction(MySQL.session(conn).status), "manual outer transaction remains open")
+    DBInterface.execute(conn, "ROLLBACK")
+    check(conn.transaction_owner === nothing, "transaction owner restored")
+
+    # An unsuccessful rollback must close the session and retain both errors.
+    caught = try
+        DBInterface.transaction(conn) do
+            DBInterface.execute(conn, "FAIL ROLLBACK")
+            throw(marker)
+        end
+        false
+    catch err
+        err isa CompositeException && length(err.exceptions) == 2 &&
+            err.exceptions[1] === marker && err.exceptions[2] isa MySQL.Error
+    end
+    check(caught, "callback and rollback errors retained")
+    check(!(isopen(conn)::Bool), "uncertain connection closed")
+    check(conn.transaction_owner === nothing, "failed cleanup restores owner")
+    return result
 end
 
 # Task bodies must be named zero-argument functions (registered via
@@ -330,29 +373,32 @@ function exercise_reaper(port::Int)::Nothing
     return nothing
 end
 
-function run_trim_workload()::Nothing
+function run_trim_workload(value::Int)::Int
     listener = TCP.listen(TCP.loopback_addr(0))
     laddr = TCP.addr(listener)::TCP.SocketAddrV4
     port = Int(laddr.port)
     SERVER_LISTENER[] = listener
     SERVER_ERROR[] = nothing
     server_task = Task(server_task_entry)
     schedule(server_task)
-    try
-        run_workload(port)
+    result = try
+        result = run_workload(port, value)
         exercise_reaper(port)
+        result
     finally
         close(listener)
         wait(server_task)
         SERVER_LISTENER[] = nothing
     end
     SERVER_ERROR[] === nothing || throw(SERVER_ERROR[])
-    return nothing
+    return result
 end
 
 function (@main)(args::Vector{String})::Cint
-    _ = args
-    run_trim_workload()
+    isempty(args) || length(args) == 2 || return 2
+    value = isempty(args) ? 9 : parse(Int, args[1])
+    expected = isempty(args) ? 19 : parse(Int, args[2])
+    run_trim_workload(value) == expected || return 1
     Core.println("mysql trim workload passed")
     return 0
 end

test/protocol/cursor_tests.jl

@@ -1080,27 +1080,31 @@ end
     end
 end
 
-@testset "transactions hold the lock; nested use is an error" begin
+@testset "transactions hold the lock across nested scopes" begin
     seen = String[]
     with_native(c -> begin
-        for _ in 1:3
-            push!(seen, expect_query(c)); send_ok(c, 1)
+        for _ in 1:9
+            sql = expect_query(c)
+            push!(seen, sql)
+            status = sql in ("COMMIT", "ROLLBACK") ? P.SERVER_STATUS_AUTOCOMMIT :
+                P.SERVER_STATUS_AUTOCOMMIT | P.SERVER_STATUS_IN_TRANS
+            send_ok(c, 1; status)
         end
-        push!(seen, expect_query(c)); send_ok(c, 1)
-        push!(seen, expect_query(c)); send_ok(c, 1)
     end) do conn
         @test DBInterface.transaction(conn) do
             DBInterface.execute(conn, "insert 1")
-            @test_throws MySQL.MySQLInterfaceError DBInterface.transaction(() -> nothing, conn)
+            @test DBInterface.transaction(() -> :nested, conn) === :nested
             @test conn.transaction_owner === current_task()
-            @test_throws MySQL.MySQLInterfaceError DBInterface.transaction(() -> nothing, conn)
+            @test DBInterface.transaction(() -> nothing, conn) === nothing
             42
         end == 42
         @test_throws ErrorException DBInterface.transaction(conn) do
             error("inside")
         end
     end
-    @test seen == ["START TRANSACTION", "insert 1", "COMMIT", "START TRANSACTION", "ROLLBACK"]
+    @test seen == ["START TRANSACTION", "insert 1", "SAVEPOINT _mysqljl_2",
+        "RELEASE SAVEPOINT _mysqljl_2", "SAVEPOINT _mysqljl_3", "RELEASE SAVEPOINT _mysqljl_3",
+        "COMMIT", "START TRANSACTION", "ROLLBACK"]
 
     seen = String[]
     with_native(c -> begin

test/protocol/runtests.jl

@@ -27,6 +27,7 @@ empty!(P.COVERAGE)
     include("native_tests.jl")
     include("cursor_tests.jl")
     include("binary_tests.jl")
+    include("transaction_tests.jl")
     include("sharedvalues_tests.jl")
     include("fuzz_tests.jl")
     include("perf_tests.jl")