fix(jdbc): propagate rollback failures and keep transaction state

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

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

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