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); + } + } +}