From e1abd18df225ade0131d67e82636f70b4ff27a3d Mon Sep 17 00:00:00 2001 From: Magnus Persson <1431685+magnusp@users.noreply.github.com> Date: Wed, 21 Oct 2026 23:19:01 +0200 Subject: [PATCH] fix(jdbc): propagate transaction end failures Share commit and rollback error handling, throw SQLException when the server rejects the transaction command, and keep the baton so callers can retry without losing transaction state. --- .../libsql/itest/NetworkToxicityIT.java | 30 ++++ .../magnusp/libsql/jdbc/LibsqlConnection.java | 56 ++++---- .../libsql/jdbc/LibsqlConnectionTest.java | 136 ++++++++++++++++++ 3 files changed, 198 insertions(+), 24 deletions(-) create mode 100644 libsql-jdbc/src/test/java/com/github/magnusp/libsql/jdbc/LibsqlConnectionTest.java diff --git a/libsql-itests/src/test/java/com/github/magnusp/libsql/itest/NetworkToxicityIT.java b/libsql-itests/src/test/java/com/github/magnusp/libsql/itest/NetworkToxicityIT.java index 4108383..0b1149a 100644 --- a/libsql-itests/src/test/java/com/github/magnusp/libsql/itest/NetworkToxicityIT.java +++ b/libsql-itests/src/test/java/com/github/magnusp/libsql/itest/NetworkToxicityIT.java @@ -131,4 +131,34 @@ void testSlicingPacketFragmentation() throws IOException, SQLException { assertThat(rs.getString(1)).isEqualTo("fragmented_packet_data_stream"); } } + + @Test + void testFailedRollbackThrowsAndKeepsTransactionForRetry() throws IOException, SQLException { + try (Connection conn = DriverManager.getConnection(getProxiedJdbcUrl(5000))) { + try (Statement st = conn.createStatement()) { + st.execute("CREATE TABLE IF NOT EXISTS toxi_rollback_test (id INTEGER PRIMARY KEY)"); + st.execute("DELETE FROM toxi_rollback_test"); + } + + conn.setAutoCommit(false); + try (Statement st = conn.createStatement()) { + st.executeUpdate("INSERT INTO toxi_rollback_test VALUES (1)"); + } + + proxy.disable(); + + assertThatThrownBy(conn::rollback).isInstanceOf(SQLException.class); + + proxy.enable(); + + conn.rollback(); + conn.setAutoCommit(true); + + try (Statement st = conn.createStatement(); + ResultSet rs = st.executeQuery("SELECT COUNT(*) FROM toxi_rollback_test")) { + assertThat(rs.next()).isTrue(); + assertThat(rs.getInt(1)).isZero(); + } + } + } } diff --git a/libsql-jdbc/src/main/java/com/github/magnusp/libsql/jdbc/LibsqlConnection.java b/libsql-jdbc/src/main/java/com/github/magnusp/libsql/jdbc/LibsqlConnection.java index b22c099..f6f6c81 100644 --- a/libsql-jdbc/src/main/java/com/github/magnusp/libsql/jdbc/LibsqlConnection.java +++ b/libsql-jdbc/src/main/java/com/github/magnusp/libsql/jdbc/LibsqlConnection.java @@ -237,19 +237,7 @@ public void commit() throws SQLException { if (autoCommit) { throw new SQLException("Cannot commit when in autoCommit mode"); } - if (baton != null) { - try { - PipelineReqBody req = new PipelineReqBody(baton, List.of( - new StreamRequest.Execute(new Stmt("COMMIT")), - StreamRequest.Close.INSTANCE - )); - PipelineRespBody resp = client.sendPipeline(req, baseUrl); - this.baton = null; - } catch (LibsqlException e) { - closeStreamSilently(); - throw LibsqlErrorMapper.toSQLException(e); - } - } + endTransaction("COMMIT"); } @Override @@ -258,18 +246,38 @@ public void rollback() throws SQLException { if (autoCommit) { throw new SQLException("Cannot rollback when in autoCommit mode"); } - if (baton != null) { - try { - PipelineReqBody req = new PipelineReqBody(baton, List.of( - new StreamRequest.Execute(new Stmt("ROLLBACK")), - StreamRequest.Close.INSTANCE - )); - client.sendPipeline(req, baseUrl); - } catch (Exception ignored) { - } finally { - this.baton = null; - } + endTransaction("ROLLBACK"); + } + + private void endTransaction(String sql) throws SQLException { + if (baton == null) { + return; + } + + PipelineRespBody resp; + try { + PipelineReqBody req = new PipelineReqBody(baton, List.of(new StreamRequest.Execute(new Stmt(sql)))); + resp = client.sendPipeline(req, baseUrl); + } catch (LibsqlException e) { + throw LibsqlErrorMapper.toSQLException(e); + } + + if (resp.results().isEmpty()) { + throw new SQLException("Empty result received from server"); + } + StreamResult result = resp.results().get(0); + if (result instanceof StreamResult.Error error) { + throw new SQLException(error.error().message()); + } + if (!(result instanceof StreamResult.Ok ok) + || !(ok.response() instanceof StreamResponse.Execute)) { + throw new SQLException("Unexpected response: " + result); + } + + if (resp.baton() != null) { + this.baton = resp.baton(); } + closeStreamSilently(); } @Override diff --git a/libsql-jdbc/src/test/java/com/github/magnusp/libsql/jdbc/LibsqlConnectionTest.java b/libsql-jdbc/src/test/java/com/github/magnusp/libsql/jdbc/LibsqlConnectionTest.java new file mode 100644 index 0000000..79d04d8 --- /dev/null +++ b/libsql-jdbc/src/test/java/com/github/magnusp/libsql/jdbc/LibsqlConnectionTest.java @@ -0,0 +1,136 @@ +package com.github.magnusp.libsql.jdbc; + +import com.github.magnusp.libsql.client.LibsqlClientConfig; +import com.github.magnusp.libsql.client.LibsqlException; +import com.github.magnusp.libsql.client.LibsqlHttpClient; +import com.github.magnusp.libsql.hrana.codec.HranaError; +import com.github.magnusp.libsql.hrana.codec.PipelineReqBody; +import com.github.magnusp.libsql.hrana.codec.PipelineRespBody; +import com.github.magnusp.libsql.hrana.codec.Stmt; +import com.github.magnusp.libsql.hrana.codec.StmtResult; +import com.github.magnusp.libsql.hrana.codec.StreamRequest; +import com.github.magnusp.libsql.hrana.codec.StreamResponse; +import com.github.magnusp.libsql.hrana.codec.StreamResult; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +import java.net.URI; +import java.sql.SQLException; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +class LibsqlConnectionTest { + + private static final LibsqlClientConfig CONFIG = new LibsqlClientConfig("http://localhost:1", null); + private static final String BATON = "baton-1"; + private static final String SERVER_ERROR = "transaction end failed on server"; + private static final String TRANSPORT_ERROR = "Connection reset by peer"; + + @ParameterizedTest + @ValueSource(strings = {"COMMIT", "ROLLBACK"}) + void serverErrorIsPropagatedAndTransactionStateIsKept(String command) throws SQLException { + AtomicInteger attempts = new AtomicInteger(); + ScriptedClient client = new ScriptedClient((request, index) -> { + if (command.equals(executedSql(request)) && attempts.incrementAndGet() == 1) { + return new PipelineRespBody(request.baton(), null, + List.of(new StreamResult.Error(new HranaError(SERVER_ERROR, "SQLITE_ERROR")))); + } + return successResponse(request); + }); + LibsqlConnection connection = connectionInTransaction(client); + + assertThatThrownBy(() -> endTransaction(connection, command)) + .isInstanceOf(SQLException.class) + .hasMessageContaining(SERVER_ERROR); + + endTransaction(connection, command); + + assertCommandRetriedOnSameStream(client, command); + } + + @ParameterizedTest + @ValueSource(strings = {"COMMIT", "ROLLBACK"}) + void transportErrorIsPropagatedAndTransactionStateIsKept(String command) throws SQLException { + AtomicInteger attempts = new AtomicInteger(); + ScriptedClient client = new ScriptedClient((request, index) -> { + if (command.equals(executedSql(request)) && attempts.incrementAndGet() == 1) { + throw new LibsqlException("Failed to execute pipeline request: " + TRANSPORT_ERROR, null, 0); + } + return successResponse(request); + }); + LibsqlConnection connection = connectionInTransaction(client); + + assertThatThrownBy(() -> endTransaction(connection, command)) + .isInstanceOf(SQLException.class) + .hasMessageContaining(TRANSPORT_ERROR); + + endTransaction(connection, command); + + assertCommandRetriedOnSameStream(client, command); + } + + private static void assertCommandRetriedOnSameStream(ScriptedClient client, String command) { + List commandRequests = client.requests.stream() + .filter(request -> command.equals(executedSql(request))) + .toList(); + assertThat(commandRequests).hasSize(2); + assertThat(commandRequests).allSatisfy( + request -> assertThat(request.baton()).isEqualTo(BATON)); + } + + private static LibsqlConnection connectionInTransaction(ScriptedClient client) throws SQLException { + LibsqlConnection connection = new LibsqlConnection(CONFIG, client); + connection.setAutoCommit(false); + connection.executeStatement(new Stmt("INSERT INTO sample VALUES (1)")); + return connection; + } + + private static void endTransaction(LibsqlConnection connection, String command) throws SQLException { + if ("COMMIT".equals(command)) { + connection.commit(); + } else { + connection.rollback(); + } + } + + private static String executedSql(PipelineReqBody request) { + if (request.requests().isEmpty() + || !(request.requests().get(0) instanceof StreamRequest.Execute execute)) { + return null; + } + return execute.stmt().sql(); + } + + private static PipelineRespBody successResponse(PipelineReqBody request) { + StreamResponse response = request.requests().get(0) instanceof StreamRequest.Close + ? StreamResponse.Close.INSTANCE + : new StreamResponse.Execute(new StmtResult(List.of(), List.of(), 0, null)); + return new PipelineRespBody(BATON, null, List.of(new StreamResult.Ok(response))); + } + + private static final class ScriptedClient extends LibsqlHttpClient { + + @FunctionalInterface + private interface Handler { + PipelineRespBody handle(PipelineReqBody request, int index); + } + + private final Handler handler; + private final List requests = new ArrayList<>(); + + private ScriptedClient(Handler handler) { + super(CONFIG); + this.handler = handler; + } + + @Override + public PipelineRespBody sendPipeline(PipelineReqBody request, URI targetUri) { + requests.add(request); + return handler.handle(request, requests.size() - 1); + } + } +}