diff --git a/src/Postgres.jl b/src/Postgres.jl index b6a472d..1cc04d3 100644 --- a/src/Postgres.jl +++ b/src/Postgres.jl @@ -906,6 +906,19 @@ end function DBInterface.close!(conn::Connection) @lock conn.lock begin if !conn.closed + # Send Terminate first, like libpq's closePGconn (sendTerminateConn + # in fe-connect.c), so the server sees an orderly close rather than + # an unexpected EOF. Best effort: the connection may already be + # broken, and close must not throw. A reentrant close, from a + # callback inside an operation, runs while that operation still + # owns the stream, and an executemany writer may be blocked on the + # socket: only closing the socket wakes it, so skip Terminate there. + if conn.lock.reentrancy_cnt == 1 && isopen(conn.socket) + try + API.writemessage(conn.socket, conn.debug, 'X') + catch + end + end close(conn.socket) conn.closed = true end diff --git a/test/isvalid_fake_server.jl b/test/isvalid_fake_server.jl index 4f7f88b..f90f746 100644 --- a/test/isvalid_fake_server.jl +++ b/test/isvalid_fake_server.jl @@ -1,8 +1,8 @@ # isvalid against a scripted server: message shapes a live PostgreSQL cannot be # made to produce on demand (a notice or parameter change while idle, a warning # followed by a bare close as on immediate shutdown, protocol garbage, a -# truncated message, a close with no bytes). Reuses the fake-server helpers -# from gssapi.jl. +# truncated message, a close with no bytes), and the Terminate message sent +# by close. Reuses the fake-server helpers from gssapi.jl. struct RecordingStyle <: Postgres.API.AbstractPostgresStyle notices::Vector{Any} @@ -12,6 +12,12 @@ RecordingStyle() = RecordingStyle(Any[], Postgres.API.Notification[]) Postgres.API.notice_callback(s::RecordingStyle, notice) = (push!(s.notices, notice); nothing) Postgres.API.notification_callback(s::RecordingStyle, n) = (push!(s.notifications, n); nothing) +# closes the connection from inside the notice callback +struct ClosingStyle <: Postgres.API.AbstractPostgresStyle + conn::Ref{Any} +end +Postgres.API.notice_callback(s::ClosingStyle, notice) = (close(s.conn[]); nothing) + fields_msg(code::Char, pairs::Pair{Char, String}...) = pgmsg(code, vcat([vcat(UInt8(k), Vector{UInt8}(v), 0x00) for (k, v) in pairs]..., 0x00)) notice_msg(severity, sqlstate, text) = fields_msg('N', 'S' => severity, 'V' => severity, 'C' => sqlstate, 'M' => text) @@ -85,6 +91,51 @@ function test_isvalid_fake_server() end end + @testset "close sends Terminate" begin + received = Tuple{Char, Vector{UInt8}}[] + with_fake_server(serve_idle(sock -> begin + push!(received, read_message(sock)) + eof(sock) || error("fake server: bytes after Terminate") + end)) do port, accepted + conn = fake_connection(port, RecordingStyle()) + close(conn) + @test !isopen(conn) + close(conn) # closing again is a no-op + @test timedwait(() -> !isempty(received), 5.0) === :ok + @test received == [('X', UInt8[])] + end + end + + @testset "close does not throw when Terminate cannot be written" begin + with_fake_server(serve_idle(sock -> begin + eof(sock) || error("fake server: bytes after a failed Terminate") + end)) do port, accepted + conn = fake_connection(port, RecordingStyle()) + # a write deadline in the past makes the Terminate write fail + Postgres.Reseau.TCP.set_write_deadline!(conn.socket.transport, 1) + @test close(conn) === nothing + @test !isopen(conn) + end + end + + @testset "close from a callback inside an operation sends no Terminate" begin + received = UInt8[] + done = Ref(false) + with_fake_server(serve_idle(sock -> begin + send(sock, notice_msg("WARNING", "01000", "close from the callback")) + while !eof(sock); push!(received, read(sock, UInt8)); end + done[] = true + end)) do port, accepted + style = ClosingStyle(Ref{Any}(nothing)) + conn = fake_connection(port, style) + style.conn[] = conn + # isvalid runs the notice callback while it holds the connection lock + @test timedwait(() -> (Postgres.isvalid(conn); !isopen(conn)), 5.0) === :ok + @test timedwait(() -> done[], 5.0) === :ok + @test isempty(received) + end + end + @testset "a message that cannot occur on an idle session is dead" begin for (label, bytes, then_close) in ( ("unexpected ReadyForQuery", pgmsg('Z', [UInt8('I')]), false),