diff --git a/docs/src/index.md b/docs/src/index.md index 6d9d928..44fab20 100644 --- a/docs/src/index.md +++ b/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)`. diff --git a/src/connection.jl b/src/connection.jl index 1c0e6b8..b66e2cc 100644 --- a/src/connection.jl +++ b/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 diff --git a/test/mysql_trim_workload.jl b/test/mysql_trim_workload.jl index 766a085..340ef5b 100644 --- a/test/mysql_trim_workload.jl +++ b/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,7 +373,7 @@ 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) @@ -338,21 +381,24 @@ function run_trim_workload()::Nothing 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 diff --git a/test/protocol/cursor_tests.jl b/test/protocol/cursor_tests.jl index 17b20b8..f06a161 100644 --- a/test/protocol/cursor_tests.jl +++ b/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 diff --git a/test/protocol/runtests.jl b/test/protocol/runtests.jl index 02c5733..5a8e3ed 100644 --- a/test/protocol/runtests.jl +++ b/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") diff --git a/test/protocol/transaction_tests.jl b/test/protocol/transaction_tests.jl new file mode 100644 index 0000000..f80f0ef --- /dev/null +++ b/test/protocol/transaction_tests.jl @@ -0,0 +1,137 @@ +@testset "Transaction scope failures and ownership" begin + in_trans = P.SERVER_STATUS_AUTOCOMMIT | P.SERVER_STATUS_IN_TRANS + marker = ErrorException("callback failed") + + @testset "Savepoint creation failure leaves caller transaction intact" begin + with_native(c -> begin + @test expect_query(c) == "START TRANSACTION" + send_ok(c, 1; status=in_trans) + @test startswith(expect_query(c), "SAVEPOINT _mysqljl_") + send_err(c, 1, 1205, "savepoint failed") + @test expect_query(c) == "ROLLBACK" + send_ok(c, 1) + end) do conn + DBInterface.execute(conn, "START TRANSACTION") + called = Ref(false) + err = try + DBInterface.transaction(conn) do + called[] = true + end + catch err + err + end + @test err isa MySQL.Error && err.errno == 1205 + @test !called[] + @test conn.transaction_owner === nothing + @test isopen(conn) + DBInterface.execute(conn, "ROLLBACK") + end + end + + for nested in (false, true), cleanup in (:rollback, :release) + !nested && cleanup === :release && continue + @testset "Failed $cleanup in nested=$nested closes the uncertain session" begin + with_native(c -> begin + @test expect_query(c) == "START TRANSACTION" + send_ok(c, 1; status=in_trans) + name = if nested + sql = expect_query(c) + @test startswith(sql, "SAVEPOINT _mysqljl_") + send_ok(c, 1; status=in_trans) + split(sql)[2] + else + nothing + end + @test expect_query(c) == (nested ? "ROLLBACK TO SAVEPOINT $name" : "ROLLBACK") + if cleanup === :release + send_ok(c, 1; status=in_trans) + @test expect_query(c) == "RELEASE SAVEPOINT $name" + end + send_err(c, 1, 1305, "cleanup failed") + end) do conn + nested && DBInterface.execute(conn, "START TRANSACTION") + err = try + DBInterface.transaction(() -> throw(marker), conn) + catch err + err + end + @test err isa CompositeException + if err isa CompositeException + @test length(err.exceptions) == 2 + @test err.exceptions[1] === marker + @test err.exceptions[2] isa MySQL.Error + @test err.exceptions[2].errno == 1305 + end + @test !isopen(conn) + @test conn.transaction_owner === nothing + @test_throws ErrorException MySQL.ping(conn) + end + end + end + + @testset "Failed commit retains its error after successful rollback" begin + with_native(c -> begin + @test expect_query(c) == "START TRANSACTION" + send_ok(c, 1; status=in_trans) + @test expect_query(c) == "COMMIT" + send_err(c, 1, 1205, "commit failed") + @test expect_query(c) == "ROLLBACK" + send_ok(c, 1) + _, cmd, _ = read_command(c) + @test cmd == P.COM_PING + send_ok(c, 1) + end) do conn + err = try + DBInterface.transaction(() -> :value, conn) + catch err + err + end + @test err isa MySQL.Error && err.errno == 1205 + @test conn.transaction_owner === nothing + @test MySQL.ping(conn) + end + end + + @testset "Ended outer scope cannot run another callback" begin + with_native(c -> begin + @test expect_query(c) == "START TRANSACTION" + send_ok(c, 1; status=in_trans) + @test expect_query(c) == "COMMIT" + send_ok(c, 1) + @test expect_query(c) == "COMMIT" + send_ok(c, 1) + end) do conn + called = Ref(false) + DBInterface.transaction(conn) do + DBInterface.execute(conn, "COMMIT") + @test_throws MySQL.MySQLInterfaceError DBInterface.transaction(conn) do + called[] = true + end + @test !called[] + @test conn.transaction_owner === current_task() + end + @test conn.transaction_owner === nothing + end + end + + @testset "Pending response is drained before choosing a scope" begin + with_native(c -> begin + @test expect_query(c) == "START TRANSACTION" + send_ok(c, 1; status=in_trans) + @test expect_query(c) == "SELECT 1" + send_resultset(c, 1, [coldef("n"; type=P.MYSQL_TYPE_LONG)], [UInt8[0x01, 0x31]]; + status=P.SERVER_STATUS_AUTOCOMMIT) + # The previous terminator ended the transaction; don't use a stale savepoint. + @test expect_query(c) == "START TRANSACTION" + send_ok(c, 1; status=in_trans) + @test expect_query(c) == "COMMIT" + send_ok(c, 1) + end) do conn + DBInterface.execute(conn, "START TRANSACTION") + cursor = DBInterface.execute(conn, "SELECT 1"; mysql_store_result=false) + @test P.in_transaction(MySQL.session(conn).status) + @test DBInterface.transaction(() -> 19, conn) == 19 + @test_throws P.ProtocolError iterate(cursor) + end + end +end diff --git a/test/transaction_workflow.jl b/test/transaction_workflow.jl new file mode 100644 index 0000000..5d3cd12 --- /dev/null +++ b/test/transaction_workflow.jl @@ -0,0 +1,90 @@ +function run_transaction_workflow(conn) + @testset "Transaction scopes preserve enclosing work" begin + DBInterface.execute(conn, "CREATE TEMPORARY TABLE transaction_scopes (id INT PRIMARY KEY) ENGINE=InnoDB") + ids() = collect(Tables.columntable(DBInterface.execute(conn, "SELECT id FROM transaction_scopes ORDER BY id")).id) + clear() = DBInterface.execute(conn, "DELETE FROM transaction_scopes") + marker = ErrorException("transaction callback failed") + try + for fail in (false, true) + clear() + DBInterface.execute(conn, "START TRANSACTION") + DBInterface.execute(conn, "INSERT INTO transaction_scopes VALUES (1)") + result = try + DBInterface.transaction(conn) do + DBInterface.execute(conn, "INSERT INTO transaction_scopes VALUES (2)") + fail && throw(marker) + :result + end + catch err + err + end + @test result === (fail ? marker : :result) + @test ids() == (fail ? [1] : [1, 2]) + @test MySQL.Protocol.in_transaction(MySQL.session(conn).status) + @test conn.transaction_owner === nothing + DBInterface.execute(conn, "ROLLBACK") + @test isempty(ids()) + end + + @test DBInterface.transaction(conn) do + DBInterface.execute(conn, "INSERT INTO transaction_scopes VALUES (1)") + result = try + DBInterface.transaction(conn) do + DBInterface.execute(conn, "INSERT INTO transaction_scopes VALUES (2)") + @test DBInterface.transaction(() -> 17, conn) == 17 + throw(marker) + end + catch err + err + end + @test result === marker + @test conn.transaction_owner === current_task() + @test ids() == [1] + DBInterface.transaction(conn) do + DBInterface.execute(conn, "INSERT INTO transaction_scopes VALUES (3)") + end + :outer_result + end === :outer_result + @test ids() == [1, 3] + @test conn.transaction_owner === nothing + + clear() + stmt = DBInterface.prepare(conn, "INSERT INTO transaction_scopes VALUES (?)") + try + @test DBInterface.transaction(conn) do + DBInterface.execute(stmt, (1,)) + @test_throws MySQL.StmtError DBInterface.executemany(stmt, ([2, 2],)) + @test ids() == [1] + @test DBInterface.executemany(stmt, ([3, 4],)) === nothing + :batch_result + end === :batch_result + @test ids() == [1, 3, 4] + @test_throws ErrorException DBInterface.transaction(conn) do + DBInterface.executemany(stmt, ([5, 6],)) + throw(marker) + end + @test ids() == [1, 3, 4] + finally + DBInterface.close!(stmt) + end + + clear() + DBInterface.execute(conn, "SET autocommit=0") + try + DBInterface.execute(conn, "INSERT INTO transaction_scopes VALUES (7)") + @test DBInterface.transaction(conn) do + DBInterface.execute(conn, "INSERT INTO transaction_scopes VALUES (8)") + :implicit_outer + end === :implicit_outer + DBInterface.execute(conn, "ROLLBACK") + @test isempty(ids()) + finally + DBInterface.execute(conn, "ROLLBACK") + DBInterface.execute(conn, "SET autocommit=1") + end + @test MySQL.ping(conn) + finally + DBInterface.execute(conn, "DROP TEMPORARY TABLE transaction_scopes") + end + end +end diff --git a/test/user_workflow.jl b/test/user_workflow.jl index e74606f..44bbca6 100644 --- a/test/user_workflow.jl +++ b/test/user_workflow.jl @@ -1,4 +1,6 @@ # Shared end-to-end API check, run on both live server lanes. +include("transaction_workflow.jl") + function run_user_workflow(conn) @testset "DBInterface and Tables user workflow" begin DBInterface.execute(conn, "CREATE DATABASE IF NOT EXISTS workflow_review") @@ -64,6 +66,7 @@ function run_user_workflow(conn) @test only(Tables.columntable(DBInterface.execute(conn, "SELECT COUNT(*) AS n FROM items")).n) == 3 @test_throws MySQL.StmtError DBInterface.execute(conn, "INSERT INTO items (id) VALUES (?)", (1,)) @test MySQL.ping(conn) + run_transaction_workflow(conn) # Five columns cross the old 4096-parameter failure at the default batch size. source = (id=collect(1:1005), label=fill("loaded", 1005), amount=fill("1.250000", 1005), stamp=fill(DateTime(2026, 1, 2), 1005), data=fill(UInt8[42], 1005))