From 1397a7a3d1f2b27cc25a2b4e3cb03900579d2a7b Mon Sep 17 00:00:00 2001 From: Jacob Quinn Date: Fri, 7 Aug 2026 08:43:37 -0600 Subject: [PATCH 1/5] Fix @transaction early-return under nesting, user catches, and tasks MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The return-rewrite threw an untagged TransactionReturn marker, so the dynamically nearest @transaction expansion always intercepted it: - a return inside a NESTED @transaction committed only the inner savepoint, and the inner expansion's own plain `return` then skipped every enclosing commit — all levels' work was silently rolled back and the connection was left inside the outer transaction - a user try/catch inside the body swallowed the marker and returned the catch's value instead of the intended return value, silently - a return inside Threads.@spawn / @async in the body was rewritten too, so the task threw the marker instead of producing its value Each expansion now tags its markers with a compile-time token. A catch that receives a foreign marker commits its own level and keeps unwinding to the owning expansion, so an early return commits every enclosing level and returns exactly once. User catch blocks get a guard injected that rethrows the marker (a private type no handler can mean to catch). Task-forming macros are excluded from the rewrite, matching the existing exclusion of closures. break/continue — which bypass both the commit and any catch — now commit via a finally, making every non-exceptional exit consistent: only a thrown exception rolls back. Documented in the docstring. Regression tests cover nested return, both catch shapes, @spawn, break, continue, and recursive re-entry of the same expansion; removing the fix fails six of them plus downstream testsets poisoned by the stuck-open transaction. Co-Authored-By: Claude Fable 5 --- src/Postgres.jl | 78 +++++++++++++++++++++++++++++++++++++----- test/runtests.jl | 89 ++++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 158 insertions(+), 9 deletions(-) diff --git a/src/Postgres.jl b/src/Postgres.jl index cbba5ba..d5dfa39 100644 --- a/src/Postgres.jl +++ b/src/Postgres.jl @@ -1218,38 +1218,84 @@ function DBInterface.transaction(f::F, conn::Connection) where {F} end end +# `token` identifies the @transaction expansion that rewrote the `return`. +# Without it, a lexically nested @transaction intercepts the outer expansion's +# marker, commits only its own savepoint, and its plain `return` then skips +# every enclosing commit — silently rolling back all levels. struct TransactionReturn{T} <: Exception + token::Symbol value::T end -function rewrite_transaction_returns(expr) +# Macros that wrap their body in a closure or task: a `return` inside them +# belongs to that closure (it becomes the task's result), so it must not be +# rewritten into a transaction-return marker. +const _TASK_MACROS = (Symbol("@spawn"), Symbol("@async"), Symbol("@task"), + Symbol("@threads"), Symbol("@distributed")) + +_macro_name(x) = x isa Symbol ? x : + x isa GlobalRef ? x.name : + (x isa Expr && x.head === :. && x.args[2] isa QuoteNode) ? x.args[2].value : + nothing + +function rewrite_transaction_returns(expr, token::Symbol) expr isa Expr || return expr if expr.head === :return - value = isempty(expr.args) ? nothing : rewrite_transaction_returns(only(expr.args)) + value = (isempty(expr.args) || expr.args[1] === nothing) ? nothing : + rewrite_transaction_returns(expr.args[1], token) marker = GlobalRef(@__MODULE__, :TransactionReturn) - return Expr(:call, GlobalRef(Core, :throw), Expr(:call, marker, value)) + return Expr(:call, GlobalRef(Core, :throw), Expr(:call, marker, QuoteNode(token), value)) elseif expr.head === :function || expr.head === :(->) || expr.head === :quote # A return in a nested function belongs to that function, not to the # scope that contains this transaction macro. return expr + elseif expr.head === :macrocall && _macro_name(expr.args[1]) in _TASK_MACROS + return expr + elseif expr.head === :try + return _rewrite_transaction_try(expr, token) + end + return Expr(expr.head, map(a -> rewrite_transaction_returns(a, token), expr.args)...) +end + +# A user `catch` inside the body would intercept the transaction-return marker +# (it is thrown as an exception) and silently produce the catch's value instead +# of returning. The marker is a private type no user handler can mean to catch, +# so re-throwing it at the top of every user catch is always correct. +function _rewrite_transaction_try(expr::Expr, token::Symbol) + args = Any[rewrite_transaction_returns(a, token) for a in expr.args] + if length(args) >= 3 && args[3] !== false + var = args[2] + if var === false + var = gensym(:transaction_err) + args[2] = var + end + marker = GlobalRef(@__MODULE__, :TransactionReturn) + guard = Expr(:&&, Expr(:call, GlobalRef(Core, :isa), var, marker), + Expr(:call, GlobalRef(Base, :rethrow))) + args[3] = Expr(:block, guard, args[3]) end - return Expr(expr.head, map(rewrite_transaction_returns, expr.args)...) + return Expr(:try, args...) end """ Postgres.@transaction conn expr -Run `expr` inside a transaction: committed if it completes, rolled back if it -throws. Evaluates to `expr`'s value. +Run `expr` inside a transaction. Any non-exceptional exit commits: normal +completion, `return` (which then returns from the enclosing function), +`break`, or `continue`. Only a thrown exception rolls back. Evaluates to +`expr`'s value. Nested `@transaction` blocks use savepoints, and an early +`return` commits every enclosing level. """ macro transaction(conn, expr) - body = rewrite_transaction_returns(expr) + token = gensym(:transaction_return) + body = rewrite_transaction_returns(expr, token) quote # bind once: the connection expression may have side effects # (`@transaction acquire(pool) ...` would otherwise take a different # connection for the BEGIN, the COMMIT and the ROLLBACK) local c = $(esc(conn)) local success = false + local completed = false start_transaction(c) try local result @@ -1257,18 +1303,32 @@ macro transaction(conn, expr) result = $(esc(body)) catch err if err isa TransactionReturn + # an early return is the success path for every enclosing + # transaction level: commit this level either way, then + # return here only if this expansion owns the marker — + # otherwise keep unwinding to the owning expansion commit(c) success = true - return err.value + completed = true + err.token === $(QuoteNode(token)) && return err.value end rethrow() end commit(c) success = true + completed = true result catch - !success && rollback_for_failed_transaction!(c) + if !success + rollback_for_failed_transaction!(c) + end + completed = true rethrow() + finally + # break/continue exit the block without passing the commit above, + # any catch, or a return: a deliberate non-exceptional exit, so it + # commits like the others + completed || commit(c) end end end diff --git a/test/runtests.jl b/test/runtests.jl index 7f05410..cfd5239 100644 --- a/test/runtests.jl +++ b/test/runtests.jl @@ -1494,6 +1494,95 @@ end @test early_return(conn) === :early @test !Postgres.in_transaction(conn) @test only(Tables.rowtable(DBInterface.execute(conn, "SELECT count(*)::int AS n FROM macro_test"))).n == 3 + + # An early return from a NESTED @transaction must commit + # every level — the inner expansion must not intercept the + # outer's marker and skip the outer commit — and must not + # leave the connection inside a transaction. + nested_return = function(c) + Postgres.@transaction c begin + DBInterface.execute(c, "INSERT INTO macro_test (value) VALUES (10)") + Postgres.@transaction c begin + DBInterface.execute(c, "INSERT INTO macro_test (value) VALUES (11)") + return :nested_early + end + end + return :late + end + @test nested_return(conn) === :nested_early + @test !Postgres.in_transaction(conn) + @test only(Tables.rowtable(DBInterface.execute(conn, "SELECT count(*)::int AS n FROM macro_test WHERE value IN (10, 11)"))).n == 2 + + # A user catch inside the body must not intercept the + # return marker and turn the return into its own value. + catch_return = function(c) + Postgres.@transaction c begin + try + DBInterface.execute(c, "INSERT INTO macro_test (value) VALUES (12)") + return :from_try + catch + return :from_catch + end + end + return :late + end + @test catch_return(conn) === :from_try + @test !Postgres.in_transaction(conn) + catch_var_return = function(c) + Postgres.@transaction c begin + try + return :from_try2 + catch err + return err + end + end + end + @test catch_var_return(conn) === :from_try2 + + # A return inside a task-forming macro is that task's + # result, not a transaction return. + spawn_return = function(c) + Postgres.@transaction c begin + t = Threads.@spawn begin + return :task_value + end + fetch(t) + end + end + @test spawn_return(conn) === :task_value + @test !Postgres.in_transaction(conn) + + # break and continue are deliberate non-exceptional exits: + # they commit, like return, and leave no transaction open. + for _ in 1:1 + Postgres.@transaction conn begin + DBInterface.execute(conn, "INSERT INTO macro_test (value) VALUES (13)") + break + end + end + @test !Postgres.in_transaction(conn) + @test only(Tables.rowtable(DBInterface.execute(conn, "SELECT count(*)::int AS n FROM macro_test WHERE value = 13"))).n == 1 + for _ in 1:2 + Postgres.@transaction conn begin + DBInterface.execute(conn, "INSERT INTO macro_test (value) VALUES (14)") + continue + end + end + @test !Postgres.in_transaction(conn) + @test only(Tables.rowtable(DBInterface.execute(conn, "SELECT count(*)::int AS n FROM macro_test WHERE value = 14"))).n == 2 + + # Recursion re-enters the SAME expansion: an inner frame's + # return exits only that frame, and each frame commits. + recursive_txn = function f(c, n) + Postgres.@transaction c begin + DBInterface.execute(c, raw"INSERT INTO macro_test (value) VALUES ($1)", (100 + n,)) + n == 0 && return :bottom + f(c, n - 1) + end + end + @test recursive_txn(conn, 2) === :bottom + @test !Postgres.in_transaction(conn) + @test only(Tables.rowtable(DBInterface.execute(conn, "SELECT count(*)::int AS n FROM macro_test WHERE value IN (100, 101, 102)"))).n == 3 end @testset "Nested Transactions" begin From eb9b4f1b1b5598d399a97170c81c24f3b3ee4309 Mon Sep 17 00:00:00 2001 From: Jacob Quinn Date: Fri, 7 Aug 2026 09:06:28 -0600 Subject: [PATCH 2/5] Exclude short-form function definitions from the @transaction rewrite MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit h(x) = ... parses as :(=) with a call-shaped left-hand side, not as :function, so the rewrite's closure exclusion missed it: a return inside a local short-form helper defined in the @transaction body was rewritten into a transaction-return marker. Calling such a helper silently early-returned the ENCLOSING function with the helper's internal value (committing on the way out), and a helper that escaped the block threw a raw TransactionReturn at its caller with no expansion active to catch it. All short-form shapes are skipped (plain, ::T return-type, where-clauses, qualified names), while ordinary assignments whose right-hand side contains a return are still rewritten. Also adds @spawnat to the task-macro skip list — same bug class as @spawn/@async, verified to wrap the marker in a RemoteException instead of producing the task's value. Live test: a short-form helper with an internal early return, used inside the block and after it escapes. Unit pins for every definition shape, the task macros, and the ordinary-assignment counter-cases. Removing the skip fails six of them. Co-Authored-By: Claude Fable 5 --- src/Postgres.jl | 17 +++++++++++++---- test/runtests.jl | 39 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 52 insertions(+), 4 deletions(-) diff --git a/src/Postgres.jl b/src/Postgres.jl index d5dfa39..065b853 100644 --- a/src/Postgres.jl +++ b/src/Postgres.jl @@ -1231,13 +1231,20 @@ end # belongs to that closure (it becomes the task's result), so it must not be # rewritten into a transaction-return marker. const _TASK_MACROS = (Symbol("@spawn"), Symbol("@async"), Symbol("@task"), - Symbol("@threads"), Symbol("@distributed")) + Symbol("@threads"), Symbol("@distributed"), Symbol("@spawnat")) _macro_name(x) = x isa Symbol ? x : x isa GlobalRef ? x.name : (x isa Expr && x.head === :. && x.args[2] isa QuoteNode) ? x.args[2].value : nothing +# Short-form function definitions — `h(x) = ...`, `h(x)::T = ...`, +# `h(x) where {T} = ...` — parse as `:(=)` with a call-shaped left-hand side. +# A return inside one belongs to that function, exactly like the long +# `function` form the rewrite already skips. +_is_callish_lhs(x) = x isa Expr && (x.head === :call || + ((x.head === :where || x.head === :(::)) && !isempty(x.args) && _is_callish_lhs(x.args[1]))) + function rewrite_transaction_returns(expr, token::Symbol) expr isa Expr || return expr if expr.head === :return @@ -1245,9 +1252,11 @@ function rewrite_transaction_returns(expr, token::Symbol) rewrite_transaction_returns(expr.args[1], token) marker = GlobalRef(@__MODULE__, :TransactionReturn) return Expr(:call, GlobalRef(Core, :throw), Expr(:call, marker, QuoteNode(token), value)) - elseif expr.head === :function || expr.head === :(->) || expr.head === :quote - # A return in a nested function belongs to that function, not to the - # scope that contains this transaction macro. + elseif expr.head === :function || expr.head === :(->) || expr.head === :quote || + (expr.head === :(=) && _is_callish_lhs(expr.args[1])) + # A return in a nested function (long form, arrow, or short form) + # belongs to that function, not to the scope that contains this + # transaction macro. return expr elseif expr.head === :macrocall && _macro_name(expr.args[1]) in _TASK_MACROS return expr diff --git a/test/runtests.jl b/test/runtests.jl index cfd5239..63d58e5 100644 --- a/test/runtests.jl +++ b/test/runtests.jl @@ -1571,6 +1571,45 @@ end @test !Postgres.in_transaction(conn) @test only(Tables.rowtable(DBInterface.execute(conn, "SELECT count(*)::int AS n FROM macro_test WHERE value = 14"))).n == 2 + # A return inside a short-form function defined in the body + # belongs to that function: it must not early-return the + # enclosing function, and the helper must stay callable + # after the block without leaking the marker. + local escaped_helper + shortform_result = (function(c) + Postgres.@transaction c begin + helper(x) = (x < 0 && return :neg; :pos) + escaped_helper = helper + (helper(-1), helper(1)) + end + end)(conn) + @test shortform_result == (:neg, :pos) + @test !Postgres.in_transaction(conn) + @test escaped_helper(-5) === :neg + + # unit-level pins for the rewrite skip list: short-form + # definitions in every syntactic shape, and task macros + let tok = gensym(:tok) + for def in (:(h(x) = return x), + :(h(x)::Int = return x), + :(h(x) where {T} = return x), + :(Base.getindex(a::MyT, i) = return i)) + @test Postgres.rewrite_transaction_returns(def, tok) == def + end + for taskex in (:(Threads.@spawn begin return 1 end), + :(Distributed.@spawnat 1 begin return 1 end), + :(@async begin return 1 end)) + @test Postgres.rewrite_transaction_returns(taskex, tok) == taskex + end + # ordinary assignments whose RHS contains a return ARE + # rewritten (x[i] = ..., x.f = ..., plain x = ...) + for assign in (:(x = f() && return 1), + :(x[i] = f() && return 1), + :(x.f = f() && return 1)) + @test Postgres.rewrite_transaction_returns(assign, tok) != assign + end + end + # Recursion re-enters the SAME expansion: an inner frame's # return exits only that frame, and each frame commits. recursive_txn = function f(c, n) From f026a66df535f2efc29c194751ed15f6ad34a6c5 Mon Sep 17 00:00:00 2001 From: Jacob Quinn Date: Fri, 7 Aug 2026 12:47:16 -0600 Subject: [PATCH 3/5] Skip comprehensions in the @transaction return rewrite MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `return` anywhere inside a comprehension or generator — body or iterator expression — is a lowering error in plain Julia. The rewrite turned it into a legal `throw` of the transaction-return marker, silently accepting code that would stop compiling the moment the @transaction wrapper is removed, and giving it early-return semantics it never legitimately had. Comprehension, typed-comprehension, generator, and flatten heads are now left untouched so the construct errors exactly as it does everywhere else. Nothing valid is lost: a legal comprehension cannot contain a bare `return`, and nested closures inside one were already excluded. Unit pins cover all four syntactic shapes plus the counter-case that a `return` inside an ordinary `for` loop is still rewritten. Co-Authored-By: Claude Fable 5 --- src/Postgres.jl | 8 ++++++++ test/runtests.jl | 10 ++++++++++ 2 files changed, 18 insertions(+) diff --git a/src/Postgres.jl b/src/Postgres.jl index 065b853..8fe5651 100644 --- a/src/Postgres.jl +++ b/src/Postgres.jl @@ -1258,6 +1258,14 @@ function rewrite_transaction_returns(expr, token::Symbol) # belongs to that function, not to the scope that contains this # transaction macro. return expr + elseif expr.head === :comprehension || expr.head === :typed_comprehension || + expr.head === :generator || expr.head === :flatten + # `return` anywhere inside a comprehension or generator (body or + # iterator expression) is a lowering error in plain Julia; rewriting + # it into a throw would silently legalize code that breaks the moment + # the @transaction wrapper is removed. Leave it to error as it always + # does. + return expr elseif expr.head === :macrocall && _macro_name(expr.args[1]) in _TASK_MACROS return expr elseif expr.head === :try diff --git a/test/runtests.jl b/test/runtests.jl index 63d58e5..c19dc8c 100644 --- a/test/runtests.jl +++ b/test/runtests.jl @@ -1608,6 +1608,16 @@ end :(x.f = f() && return 1)) @test Postgres.rewrite_transaction_returns(assign, tok) != assign end + # a return inside a comprehension or generator is a + # lowering error in plain Julia; the rewrite must not + # legalize it into a throw that works only while the + # @transaction wrapper is present + for comp in (:([(return i) for i in 1:3]), + :(Int[(return i) for i in 1:3]), + :(sum(x for x in (f() ? (return 1) : [1]))), + :([x for x in xs for y in (return x)])) + @test Postgres.rewrite_transaction_returns(comp, tok) == comp + end end # Recursion re-enters the SAME expansion: an inner frame's From 8f9711be8e8e98ade433781b480870204b4c2a95 Mon Sep 17 00:00:00 2001 From: Jacob Quinn Date: Fri, 7 Aug 2026 13:44:41 -0600 Subject: [PATCH 4/5] Add @fetch/@fetchfrom to the @transaction task-macro skip list MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Distributed.@fetch and @fetchfrom wrap their body in a remotely-executed thunk whose return is the fetched value, exactly like @spawnat — but they were missing from _TASK_MACROS, so a return inside one was rewritten into a transaction-return marker. Verified live: the block then throws a RemoteException wrapping the marker and rolls back, where plain Julia returns the value. Also corrects the comprehension-skip rationale in comments: a return in a comprehension/generator BODY is a lowering error (which the rewrite must not legalize), while the iterator-expression shapes lowering does accept behave correctly un-rewritten — they exit the block non-exceptionally and commit through the expansion's finally, as verified live. And documents at the token comparison that unconditional returning would be observationally equivalent today only because every enclosing expansion's finally also commits; the token check stays as the semantic guarantee. Independent adversarial verification of the three @transaction commits (61 live scenarios, plain-Julia baselines, 6 mutations against the full suite) found no other behavioral gaps. Co-Authored-By: Claude Fable 5 --- src/Postgres.jl | 23 ++++++++++++++++------- test/runtests.jl | 13 ++++++++----- 2 files changed, 24 insertions(+), 12 deletions(-) diff --git a/src/Postgres.jl b/src/Postgres.jl index 8fe5651..04b44af 100644 --- a/src/Postgres.jl +++ b/src/Postgres.jl @@ -1231,7 +1231,8 @@ end # belongs to that closure (it becomes the task's result), so it must not be # rewritten into a transaction-return marker. const _TASK_MACROS = (Symbol("@spawn"), Symbol("@async"), Symbol("@task"), - Symbol("@threads"), Symbol("@distributed"), Symbol("@spawnat")) + Symbol("@threads"), Symbol("@distributed"), Symbol("@spawnat"), + Symbol("@fetch"), Symbol("@fetchfrom")) _macro_name(x) = x isa Symbol ? x : x isa GlobalRef ? x.name : @@ -1260,11 +1261,14 @@ function rewrite_transaction_returns(expr, token::Symbol) return expr elseif expr.head === :comprehension || expr.head === :typed_comprehension || expr.head === :generator || expr.head === :flatten - # `return` anywhere inside a comprehension or generator (body or - # iterator expression) is a lowering error in plain Julia; rewriting - # it into a throw would silently legalize code that breaks the moment - # the @transaction wrapper is removed. Leave it to error as it always - # does. + # A `return` in a comprehension/generator body is a lowering error in + # plain Julia; rewriting it into a throw would silently legalize code + # that breaks the moment the @transaction wrapper is removed. The + # shapes plain lowering does accept (a return in an iterator + # expression evaluated in the enclosing scope) exit the block + # non-exceptionally and commit through the finally below, exactly as + # they behave outside the macro — so leaving the whole construct + # untouched is right in both cases. return expr elseif expr.head === :macrocall && _macro_name(expr.args[1]) in _TASK_MACROS return expr @@ -1323,7 +1327,12 @@ macro transaction(conn, expr) # an early return is the success path for every enclosing # transaction level: commit this level either way, then # return here only if this expansion owns the marker — - # otherwise keep unwinding to the owning expansion + # otherwise keep unwinding to the owning expansion. + # (Returning unconditionally would be observationally + # equivalent today because every enclosing expansion's + # finally also commits on a non-exceptional exit; the + # token check is kept as the semantic guarantee rather + # than leaning on that structural accident.) commit(c) success = true completed = true diff --git a/test/runtests.jl b/test/runtests.jl index c19dc8c..822ad25 100644 --- a/test/runtests.jl +++ b/test/runtests.jl @@ -1598,7 +1598,9 @@ end end for taskex in (:(Threads.@spawn begin return 1 end), :(Distributed.@spawnat 1 begin return 1 end), - :(@async begin return 1 end)) + :(@async begin return 1 end), + :(Distributed.@fetch begin return 1 end), + :(@fetchfrom 1 begin return 1 end)) @test Postgres.rewrite_transaction_returns(taskex, tok) == taskex end # ordinary assignments whose RHS contains a return ARE @@ -1608,10 +1610,11 @@ end :(x.f = f() && return 1)) @test Postgres.rewrite_transaction_returns(assign, tok) != assign end - # a return inside a comprehension or generator is a - # lowering error in plain Julia; the rewrite must not - # legalize it into a throw that works only while the - # @transaction wrapper is present + # a return in a comprehension/generator body is a + # lowering error in plain Julia (the rewrite must not + # legalize it), and the iterator-expression shapes + # lowering does accept behave correctly un-rewritten + # (they commit through the expansion's finally) for comp in (:([(return i) for i in 1:3]), :(Int[(return i) for i in 1:3]), :(sum(x for x in (f() ? (return 1) : [1]))), From 467596923c86b4f483d107174b8a28d4edd7f476 Mon Sep 17 00:00:00 2001 From: Jacob Quinn Date: Fri, 7 Aug 2026 14:41:04 -0600 Subject: [PATCH 5/5] Remove the @transaction return rewrite; commit non-local exits via finally MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review of the rewrite approach (PR #7 threads) proved its task-macro allowlist structurally insufficient: any third-party macro that wraps its body in a task or closure — reproduced with a minimal @local_task — had its internal returns rewritten into transaction-return markers, throwing TaskFailedException(TransactionReturn) instead of producing the task's value. No finite list of standard macros covers user-defined ones. The expansion's finally already gives plain `return` the intended semantics with no rewriting at all: a return unwinds through every enclosing expansion's finally, each committing its level exactly once, innermost first. User catches cannot intercept a plain return, closures and task macros keep their ordinary meaning untouched, and the flattened-iterator form that plain lowering accepts behaves identically wrapped or not. The marker struct, the AST walker, the try-guard injection, and both skip lists are deleted. The finally also now handles its own commit failure: it rolls back the current level before propagating (commit at savepoint depth leaves depth unchanged on failure), so every enclosing level — macro expansion or plain catch — unwinds its own. Previously a RELEASE SAVEPOINT failure during a break out of a nested level (savepoint aborted by a swallowed server error) escaped past the enclosing macro's ability to clean up, leaving the outer transaction open with its work pending. Behavioral regressions replace the deleted unit AST pins: a third-party @local_task macro, Distributed @spawnat/@fetch/@fetchfrom run locally on worker 1, plain-vs-wrapped flattened-iterator equivalence, and the nested break with an aborted savepoint (asserts the server error surfaces and nothing stays open client- or server-side; removing the finally rollback fails five assertions). Co-Authored-By: Claude Fable 5 --- Project.toml | 4 +- src/Postgres.jl | 133 +++++++++-------------------------------------- test/runtests.jl | 122 ++++++++++++++++++++++++++++++++----------- 3 files changed, 119 insertions(+), 140 deletions(-) diff --git a/Project.toml b/Project.toml index dce5a9c..09cd63e 100644 --- a/Project.toml +++ b/Project.toml @@ -22,6 +22,7 @@ Aqua = "0.8" ConcurrentUtilities = "2.1" DBInterface = "2.5" Dates = "1.10" +Distributed = "1.10" Harbor = "1" JSON = "1" MD5 = "0.2" @@ -39,9 +40,10 @@ julia = "1.10" [extras] Aqua = "4c88cf16-eb10-579e-8560-4a9242c79595" +Distributed = "8ba89e20-285c-5b6f-9357-94700520ee1b" Harbor = "af79dbb9-1a80-47ad-8928-192a4af69376" Sockets = "6462fe0b-24de-5631-8697-dd941f90decc" Test = "8dfed614-e22c-5e08-85e1-65c5234f0b40" [targets] -test = ["Aqua", "Harbor", "Sockets", "Test"] +test = ["Aqua", "Distributed", "Harbor", "Sockets", "Test"] diff --git a/src/Postgres.jl b/src/Postgres.jl index 04b44af..aaab198 100644 --- a/src/Postgres.jl +++ b/src/Postgres.jl @@ -1218,86 +1218,6 @@ function DBInterface.transaction(f::F, conn::Connection) where {F} end end -# `token` identifies the @transaction expansion that rewrote the `return`. -# Without it, a lexically nested @transaction intercepts the outer expansion's -# marker, commits only its own savepoint, and its plain `return` then skips -# every enclosing commit — silently rolling back all levels. -struct TransactionReturn{T} <: Exception - token::Symbol - value::T -end - -# Macros that wrap their body in a closure or task: a `return` inside them -# belongs to that closure (it becomes the task's result), so it must not be -# rewritten into a transaction-return marker. -const _TASK_MACROS = (Symbol("@spawn"), Symbol("@async"), Symbol("@task"), - Symbol("@threads"), Symbol("@distributed"), Symbol("@spawnat"), - Symbol("@fetch"), Symbol("@fetchfrom")) - -_macro_name(x) = x isa Symbol ? x : - x isa GlobalRef ? x.name : - (x isa Expr && x.head === :. && x.args[2] isa QuoteNode) ? x.args[2].value : - nothing - -# Short-form function definitions — `h(x) = ...`, `h(x)::T = ...`, -# `h(x) where {T} = ...` — parse as `:(=)` with a call-shaped left-hand side. -# A return inside one belongs to that function, exactly like the long -# `function` form the rewrite already skips. -_is_callish_lhs(x) = x isa Expr && (x.head === :call || - ((x.head === :where || x.head === :(::)) && !isempty(x.args) && _is_callish_lhs(x.args[1]))) - -function rewrite_transaction_returns(expr, token::Symbol) - expr isa Expr || return expr - if expr.head === :return - value = (isempty(expr.args) || expr.args[1] === nothing) ? nothing : - rewrite_transaction_returns(expr.args[1], token) - marker = GlobalRef(@__MODULE__, :TransactionReturn) - return Expr(:call, GlobalRef(Core, :throw), Expr(:call, marker, QuoteNode(token), value)) - elseif expr.head === :function || expr.head === :(->) || expr.head === :quote || - (expr.head === :(=) && _is_callish_lhs(expr.args[1])) - # A return in a nested function (long form, arrow, or short form) - # belongs to that function, not to the scope that contains this - # transaction macro. - return expr - elseif expr.head === :comprehension || expr.head === :typed_comprehension || - expr.head === :generator || expr.head === :flatten - # A `return` in a comprehension/generator body is a lowering error in - # plain Julia; rewriting it into a throw would silently legalize code - # that breaks the moment the @transaction wrapper is removed. The - # shapes plain lowering does accept (a return in an iterator - # expression evaluated in the enclosing scope) exit the block - # non-exceptionally and commit through the finally below, exactly as - # they behave outside the macro — so leaving the whole construct - # untouched is right in both cases. - return expr - elseif expr.head === :macrocall && _macro_name(expr.args[1]) in _TASK_MACROS - return expr - elseif expr.head === :try - return _rewrite_transaction_try(expr, token) - end - return Expr(expr.head, map(a -> rewrite_transaction_returns(a, token), expr.args)...) -end - -# A user `catch` inside the body would intercept the transaction-return marker -# (it is thrown as an exception) and silently produce the catch's value instead -# of returning. The marker is a private type no user handler can mean to catch, -# so re-throwing it at the top of every user catch is always correct. -function _rewrite_transaction_try(expr::Expr, token::Symbol) - args = Any[rewrite_transaction_returns(a, token) for a in expr.args] - if length(args) >= 3 && args[3] !== false - var = args[2] - if var === false - var = gensym(:transaction_err) - args[2] = var - end - marker = GlobalRef(@__MODULE__, :TransactionReturn) - guard = Expr(:&&, Expr(:call, GlobalRef(Core, :isa), var, marker), - Expr(:call, GlobalRef(Base, :rethrow))) - args[3] = Expr(:block, guard, args[3]) - end - return Expr(:try, args...) -end - """ Postgres.@transaction conn expr @@ -1305,11 +1225,14 @@ Run `expr` inside a transaction. Any non-exceptional exit commits: normal completion, `return` (which then returns from the enclosing function), `break`, or `continue`. Only a thrown exception rolls back. Evaluates to `expr`'s value. Nested `@transaction` blocks use savepoints, and an early -`return` commits every enclosing level. +`return` commits every enclosing level on its way out. + +The body keeps plain Julia semantics: a `return` inside a nested function, +closure, `do`-block, or any task-forming macro (`Threads.@spawn`, `@async`, +`Distributed.@spawnat`, third-party equivalents) belongs to that function or +task, exactly as it would outside the macro. """ macro transaction(conn, expr) - token = gensym(:transaction_return) - body = rewrite_transaction_returns(expr, token) quote # bind once: the connection expression may have side effects # (`@transaction acquire(pool) ...` would otherwise take a different @@ -1319,27 +1242,7 @@ macro transaction(conn, expr) local completed = false start_transaction(c) try - local result - try - result = $(esc(body)) - catch err - if err isa TransactionReturn - # an early return is the success path for every enclosing - # transaction level: commit this level either way, then - # return here only if this expansion owns the marker — - # otherwise keep unwinding to the owning expansion. - # (Returning unconditionally would be observationally - # equivalent today because every enclosing expansion's - # finally also commits on a non-exceptional exit; the - # token check is kept as the semantic guarantee rather - # than leaning on that structural accident.) - commit(c) - success = true - completed = true - err.token === $(QuoteNode(token)) && return err.value - end - rethrow() - end + local result = $(esc(expr)) commit(c) success = true completed = true @@ -1351,10 +1254,24 @@ macro transaction(conn, expr) completed = true rethrow() finally - # break/continue exit the block without passing the commit above, - # any catch, or a return: a deliberate non-exceptional exit, so it - # commits like the others - completed || commit(c) + # A non-exceptional, non-local exit — return, break, continue — + # reaches here without passing the commit above or the catch: + # commit this level on the way out. `return` unwinds through every + # enclosing expansion's finally, so each level commits exactly + # once, innermost first; no AST rewriting is needed, and returns + # inside closures or task-forming macros keep their plain-Julia + # meaning untouched. If the commit fails, roll back THIS level + # before propagating (commit at savepoint depth leaves the depth + # unchanged on failure), so every enclosing level — macro + # expansion or plain catch — can then unwind its own. + if !completed + try + commit(c) + catch + rollback_for_failed_transaction!(c) + rethrow() + end + end end end end diff --git a/test/runtests.jl b/test/runtests.jl index 822ad25..239dd79 100644 --- a/test/runtests.jl +++ b/test/runtests.jl @@ -1,6 +1,7 @@ using Test using Aqua using Dates +using Distributed using UUIDs using DBInterface using Tables @@ -70,6 +71,17 @@ struct Int8Row s::String end +# A third-party-style task macro the driver has never heard of: it wraps its +# body in a Task and fetches it. @transaction must leave the body's `return` +# with its plain meaning (the task's result) — no allowlist involved. +macro local_task(body) + quote + local t = Task(() -> $(esc(body))) + schedule(t) + fetch(t) + end +end + # user-IO failure injection for the COPY hardening tests struct ThrowingSource <: IO end Base.eof(::ThrowingSource) = false @@ -1587,41 +1599,89 @@ end @test !Postgres.in_transaction(conn) @test escaped_helper(-5) === :neg - # unit-level pins for the rewrite skip list: short-form - # definitions in every syntactic shape, and task macros - let tok = gensym(:tok) - for def in (:(h(x) = return x), - :(h(x)::Int = return x), - :(h(x) where {T} = return x), - :(Base.getindex(a::MyT, i) = return i)) - @test Postgres.rewrite_transaction_returns(def, tok) == def - end - for taskex in (:(Threads.@spawn begin return 1 end), - :(Distributed.@spawnat 1 begin return 1 end), - :(@async begin return 1 end), - :(Distributed.@fetch begin return 1 end), - :(@fetchfrom 1 begin return 1 end)) - @test Postgres.rewrite_transaction_returns(taskex, tok) == taskex + # The body keeps plain Julia semantics with no AST rewrite, + # so a return inside ANY closure-forming construct behaves + # exactly as it does outside the macro — including + # third-party task macros no allowlist could cover. + local_task_result = (function(c) + Postgres.@transaction c begin + DBInterface.execute(c, "INSERT INTO macro_test (value) VALUES (20)") + @local_task begin + return :local_task_value + end end - # ordinary assignments whose RHS contains a return ARE - # rewritten (x[i] = ..., x.f = ..., plain x = ...) - for assign in (:(x = f() && return 1), - :(x[i] = f() && return 1), - :(x.f = f() && return 1)) - @test Postgres.rewrite_transaction_returns(assign, tok) != assign + end)(conn) + @test local_task_result === :local_task_value + @test !Postgres.in_transaction(conn) + @test only(Tables.rowtable(DBInterface.execute(conn, "SELECT count(*)::int AS n FROM macro_test WHERE value = 20"))).n == 1 + + # Distributed task macros: with no workers added, worker 1 + # is this process, so these run locally end to end. + for distributed_case in ( + (c) -> Postgres.@transaction(c, fetch(Distributed.@spawnat 1 begin + return :spawnat_value + end)), + (c) -> Postgres.@transaction(c, Distributed.@fetchfrom 1 begin + return :fetchfrom_value + end), + (c) -> Postgres.@transaction(c, Distributed.@fetch begin + return :fetch_value + end), + ) + val = distributed_case(conn) + @test val in (:spawnat_value, :fetchfrom_value, :fetch_value) + @test !Postgres.in_transaction(conn) + end + + # A return in a flattened-iterator expression is legal + # plain Julia: it returns from the generated per-element + # closure, so its value becomes the inner iterator (an Int + # yields itself once) and the enclosing function continues. + # Wrapped in @transaction the behavior must be identical, + # and the block commits on normal completion. + plain_flatten = function() + vals = [x for x in 1:2 for y in (return x)] + (:reached, vals) + end + wrapped_flatten = function(c) + vals = Postgres.@transaction c begin + DBInterface.execute(c, "INSERT INTO macro_test (value) VALUES (21)") + [x for x in 1:2 for y in (return x)] end - # a return in a comprehension/generator body is a - # lowering error in plain Julia (the rewrite must not - # legalize it), and the iterator-expression shapes - # lowering does accept behave correctly un-rewritten - # (they commit through the expansion's finally) - for comp in (:([(return i) for i in 1:3]), - :(Int[(return i) for i in 1:3]), - :(sum(x for x in (f() ? (return 1) : [1]))), - :([x for x in xs for y in (return x)])) - @test Postgres.rewrite_transaction_returns(comp, tok) == comp + (:reached, vals) + end + @test plain_flatten() == (:reached, [1, 2]) + @test wrapped_flatten(conn) == plain_flatten() + @test !Postgres.in_transaction(conn) + @test only(Tables.rowtable(DBInterface.execute(conn, "SELECT count(*)::int AS n FROM macro_test WHERE value = 21"))).n == 1 + + # A commit that fails in the finally (break out of a nested + # level whose savepoint was aborted by a swallowed server + # error) must roll back ITS level before propagating, so + # every enclosing level can unwind its own — nothing may be + # left open, client- or server-side. + nested_break_err = try + for _ in 1:1 + Postgres.@transaction conn begin + DBInterface.execute(conn, "INSERT INTO macro_test (value) VALUES (22)") + Postgres.@transaction conn begin + try + DBInterface.execute(conn, "SELECT 1/0") + catch + # swallowed: the savepoint is now aborted + end + break + end + end end + nothing + catch e + e end + @test nested_break_err isa Postgres.API.Error + @test !Postgres.in_transaction(conn) + @test !conn.server_in_transaction + @test isempty(Tables.rowtable(DBInterface.execute(conn, "SELECT * FROM macro_test WHERE value = 22"))) # Recursion re-enters the SAME expansion: an inner frame's # return exits only that frame, and each frame commits.