diff --git a/libsql-itests/src/test/java/com/github/magnusp/libsql/itest/TransactionLifecycleIT.java b/libsql-itests/src/test/java/com/github/magnusp/libsql/itest/TransactionLifecycleIT.java new file mode 100644 index 0000000..525abec --- /dev/null +++ b/libsql-itests/src/test/java/com/github/magnusp/libsql/itest/TransactionLifecycleIT.java @@ -0,0 +1,126 @@ +package com.github.magnusp.libsql.itest; + +import com.github.magnusp.libsql.jdbc.LibsqlConnection; +import org.junit.jupiter.api.Test; + +import java.sql.Connection; +import java.sql.DriverManager; +import java.sql.SQLException; +import java.sql.Statement; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; + +import static org.assertj.core.api.Assertions.assertThat; + +public class TransactionLifecycleIT extends SqldContainerBase { + + private String getDriverJdbcUrl() { + return "jdbc:libsql:" + getHttpUrl(); + } + + @Test + void deferredTransactionBeginsEagerlyWithoutHoldingAWriteLock() throws Exception { + String url = getDriverJdbcUrl(); + try (Connection lockHolder = DriverManager.getConnection(url); + Connection writer = DriverManager.getConnection(url)) { + createTable(lockHolder, "deferred_lock"); + + lockHolder.setAutoCommit(false); + + ExecutorService executor = Executors.newSingleThreadExecutor(); + try { + Future insert = executor.submit(() -> insertRow(writer, "deferred_lock", 1)); + assertThat(insert.get(5, TimeUnit.SECONDS)).isEqualTo(1); + } finally { + executor.shutdownNow(); + } + } + } + + @Test + void immediateTransactionBeginsEagerlyAndHoldsTheWriteLock() throws Exception { + String url = getDriverJdbcUrl(); + try (Connection lockHolder = DriverManager.getConnection(url)) { + createTable(lockHolder, "immediate_lock"); + LibsqlConnection libsql = lockHolder.unwrap(LibsqlConnection.class); + libsql.setTransactionMode(LibsqlConnection.TransactionMode.IMMEDIATE); + + lockHolder.setAutoCommit(false); + + assertWriteIsBlockedUntilReleased(url, lockHolder, "immediate_lock", 1); + } + } + + @Test + void commitBeginsTheNextTransactionAndKeepsTheConfiguredLock() throws Exception { + String url = getDriverJdbcUrl(); + try (Connection lockHolder = DriverManager.getConnection(url); + Statement statement = lockHolder.createStatement()) { + createTable(lockHolder, "commit_rebegin"); + LibsqlConnection libsql = lockHolder.unwrap(LibsqlConnection.class); + libsql.setTransactionMode(LibsqlConnection.TransactionMode.IMMEDIATE); + lockHolder.setAutoCommit(false); + statement.executeUpdate("INSERT INTO commit_rebegin (id) VALUES (100)"); + + lockHolder.commit(); + + assertWriteIsBlockedUntilReleased(url, lockHolder, "commit_rebegin", 1); + } + } + + private static void assertWriteIsBlockedUntilReleased( + String url, Connection lockHolder, String table, int id) throws Exception { + ExecutorService executor = Executors.newSingleThreadExecutor(); + try { + Future insert = executor.submit(() -> { + try (Connection writer = DriverManager.getConnection(url)) { + return insertRow(writer, table, id); + } catch (SQLException e) { + return e; + } + }); + + Object resultWhileLocked; + try { + resultWhileLocked = insert.get(300, TimeUnit.MILLISECONDS); + assertThat(resultWhileLocked) + .as("writer must not succeed while the write lock is held") + .isNotEqualTo(1); + } catch (TimeoutException blocked) { + resultWhileLocked = null; + } + + lockHolder.setAutoCommit(true); + + if (insert.isDone()) { + Object completed = insert.get(); + if (completed instanceof SQLException) { + try (Connection writer = DriverManager.getConnection(url)) { + assertThat(insertRow(writer, table, id)).isEqualTo(1); + } + } else { + assertThat(completed).isEqualTo(1); + } + } else { + assertThat(insert.get(20, TimeUnit.SECONDS)).isEqualTo(1); + } + } finally { + executor.shutdownNow(); + } + } + + private static void createTable(Connection connection, String table) throws SQLException { + try (Statement statement = connection.createStatement()) { + statement.execute("CREATE TABLE " + table + " (id INTEGER PRIMARY KEY)"); + } + } + + private static int insertRow(Connection connection, String table, int id) throws SQLException { + try (Statement statement = connection.createStatement()) { + return statement.executeUpdate("INSERT INTO " + table + " (id) VALUES (" + id + ")"); + } + } +} 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 f6f6c81..b34008a 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 @@ -46,6 +46,35 @@ public class LibsqlConnection implements Connection { private boolean readOnly = false; private int transactionIsolation = TRANSACTION_SERIALIZABLE; private boolean closed = false; + private TransactionMode transactionMode = TransactionMode.DEFERRED; + + /** + * Transaction begin mode, matching xerial sqlite-jdbc's + * {@code SQLiteConfig.TransactionMode}. + */ + public enum TransactionMode { + DEFERRED("BEGIN"), + IMMEDIATE("BEGIN IMMEDIATE"), + EXCLUSIVE("BEGIN EXCLUSIVE"); + + private final String beginSql; + + TransactionMode(String beginSql) { + this.beginSql = beginSql; + } + + String beginSql() { + return beginSql; + } + + public static TransactionMode from(String mode) { + try { + return valueOf(mode.toUpperCase()); + } catch (RuntimeException e) { + throw new IllegalArgumentException("Unknown transaction mode: " + mode, e); + } + } + } // Interactive stream state private String baton = null; @@ -159,15 +188,33 @@ private int[] extractBatchResults(PipelineRespBody resp, int expectedSize) throw private void ensureTransactionStarted() throws SQLException { if (baton == null) { - // Mitigation for Bug 2: Read-write transactions start with BEGIN IMMEDIATE - String beginSql = readOnly ? "BEGIN DEFERRED" : "BEGIN IMMEDIATE"; - PipelineReqBody req = new PipelineReqBody(null, List.of(new StreamRequest.Execute(new Stmt(beginSql)))); - PipelineRespBody resp = client.sendPipeline(req, baseUrl); - updateStreamState(resp); - if (resp.results().isEmpty() || resp.results().get(0) instanceof StreamResult.Error) { - closeStreamSilently(); - throw new SQLException("Failed to begin transaction"); - } + beginTransaction(); + } + } + + private void beginTransaction() throws SQLException { + PipelineRespBody resp; + try { + PipelineReqBody req = new PipelineReqBody(null, + List.of(new StreamRequest.Execute(new Stmt(transactionMode.beginSql())))); + resp = client.sendPipeline(req, baseUrl); + } catch (LibsqlException e) { + throw LibsqlErrorMapper.toSQLException(e); + } + updateStreamState(resp); + if (resp.results().isEmpty()) { + closeStreamSilently(); + throw new SQLException("Empty result received from server"); + } + StreamResult result = resp.results().get(0); + if (result instanceof StreamResult.Error error) { + closeStreamSilently(); + throw new SQLException(error.error().message()); + } + if (!(result instanceof StreamResult.Ok ok) + || !(ok.response() instanceof StreamResponse.Execute)) { + closeStreamSilently(); + throw new SQLException("Unexpected response: " + result); } } @@ -214,15 +261,33 @@ public String nativeSQL(String sql) throws SQLException { return sql; } + /** + * Matches xerial sqlite-jdbc 3.53.4.0: + * + * + * + *

Use {@link #setTransactionMode(TransactionMode)} to select {@code IMMEDIATE} or {@code + * EXCLUSIVE}, which acquire the write lock when the transaction begins. + */ @Override public void setAutoCommit(boolean autoCommit) throws SQLException { checkClosed(); if (this.autoCommit == autoCommit) return; - if (!this.autoCommit && autoCommit) { - // Committing active transaction - commit(); + if (autoCommit) { + endTransaction("COMMIT", false); + this.autoCommit = true; + } else { + this.autoCommit = false; + beginTransaction(); } - this.autoCommit = autoCommit; } @Override @@ -237,7 +302,7 @@ public void commit() throws SQLException { if (autoCommit) { throw new SQLException("Cannot commit when in autoCommit mode"); } - endTransaction("COMMIT"); + endTransaction("COMMIT", true); } @Override @@ -246,18 +311,26 @@ public void rollback() throws SQLException { if (autoCommit) { throw new SQLException("Cannot rollback when in autoCommit mode"); } - endTransaction("ROLLBACK"); + endTransaction("ROLLBACK", true); } - private void endTransaction(String sql) throws SQLException { + private void endTransaction(String sql, boolean rebegin) throws SQLException { if (baton == null) { + if (rebegin) { + throw new SQLException("No transaction is active"); + } return; } + List requests = new ArrayList<>(2); + requests.add(new StreamRequest.Execute(new Stmt(sql))); + if (rebegin) { + requests.add(new StreamRequest.Execute(new Stmt(transactionMode.beginSql()))); + } + PipelineRespBody resp; try { - PipelineReqBody req = new PipelineReqBody(baton, List.of(new StreamRequest.Execute(new Stmt(sql)))); - resp = client.sendPipeline(req, baseUrl); + resp = client.sendPipeline(new PipelineReqBody(baton, requests), baseUrl); } catch (LibsqlException e) { throw LibsqlErrorMapper.toSQLException(e); } @@ -274,10 +347,33 @@ private void endTransaction(String sql) throws SQLException { throw new SQLException("Unexpected response: " + result); } - if (resp.baton() != null) { - this.baton = resp.baton(); + if (!rebegin) { + updateStreamState(resp); + closeStreamSilently(); + return; + } + + if (resp.results().size() < 2) { + closeStreamSilently(); + throw new SQLException("Missing response for transaction begin"); } - closeStreamSilently(); + StreamResult begin = resp.results().get(1); + if (begin instanceof StreamResult.Error error) { + if (resp.baton() != null) { + this.baton = resp.baton(); + } + closeStreamSilently(); + throw new SQLException(error.error().message()); + } + if (!(begin instanceof StreamResult.Ok beginOk) + || !(beginOk.response() instanceof StreamResponse.Execute)) { + if (resp.baton() != null) { + this.baton = resp.baton(); + } + closeStreamSilently(); + throw new SQLException("Unexpected response: " + begin); + } + updateStreamState(resp); } @Override @@ -319,6 +415,27 @@ public String getCatalog() throws SQLException { return null; } + /** + * Selects the mode used by the next transaction begin. Matches xerial sqlite-jdbc's {@code + * SQLiteConfig.setTransactionMode}; the default is {@link TransactionMode#DEFERRED}. + */ + public void setTransactionMode(TransactionMode mode) throws SQLException { + checkClosed(); + if (mode == null) { + throw new IllegalArgumentException("Transaction mode cannot be null"); + } + this.transactionMode = mode; + } + + public void setTransactionMode(String mode) throws SQLException { + setTransactionMode(TransactionMode.from(mode)); + } + + public TransactionMode getTransactionMode() throws SQLException { + checkClosed(); + return transactionMode; + } + @Override public void setTransactionIsolation(int level) throws SQLException { checkClosed(); 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 index 79d04d8..1153c5f 100644 --- 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 @@ -11,7 +11,9 @@ 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.api.Test; import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.EnumSource; import org.junit.jupiter.params.provider.ValueSource; import java.net.URI; @@ -30,6 +32,88 @@ class LibsqlConnectionTest { private static final String SERVER_ERROR = "transaction end failed on server"; private static final String TRANSPORT_ERROR = "Connection reset by peer"; + @Test + void disablingAutoCommitBeginsTransactionImmediately() throws SQLException { + ScriptedClient client = new ScriptedClient((request, index) -> successResponse(request)); + LibsqlConnection connection = new LibsqlConnection(CONFIG, client); + + connection.setAutoCommit(false); + + assertThat(client.executedStatements()).containsExactly("BEGIN"); + assertThat(client.requests.get(0).baton()).isNull(); + } + + @ParameterizedTest + @EnumSource(value = LibsqlConnection.TransactionMode.class, names = {"IMMEDIATE", "EXCLUSIVE"}) + void disablingAutoCommitUsesConfiguredTransactionMode(LibsqlConnection.TransactionMode mode) + throws SQLException { + ScriptedClient client = new ScriptedClient((request, index) -> successResponse(request)); + LibsqlConnection connection = new LibsqlConnection(CONFIG, client); + connection.setTransactionMode(mode); + + connection.setAutoCommit(false); + + String expected = switch (mode) { + case IMMEDIATE -> "BEGIN IMMEDIATE"; + case EXCLUSIVE -> "BEGIN EXCLUSIVE"; + case DEFERRED -> "BEGIN"; + }; + assertThat(client.executedStatements()).containsExactly(expected); + } + + @Test + void commitReBeginsInConfiguredTransactionMode() throws SQLException { + ScriptedClient client = new ScriptedClient((request, index) -> successResponse(request)); + LibsqlConnection connection = new LibsqlConnection(CONFIG, client); + connection.setTransactionMode("immediate"); + connection.setAutoCommit(false); + + connection.commit(); + + assertThat(client.executedStatements()) + .containsExactly("BEGIN IMMEDIATE", "COMMIT", "BEGIN IMMEDIATE"); + assertThat(connection.getTransactionMode()) + .isEqualTo(LibsqlConnection.TransactionMode.IMMEDIATE); + } + + @Test + void commitBeginsNextTransactionOnTheSameStream() throws SQLException { + ScriptedClient client = new ScriptedClient((request, index) -> successResponse(request)); + LibsqlConnection connection = new LibsqlConnection(CONFIG, client); + connection.setAutoCommit(false); + + connection.commit(); + + assertThat(client.executedStatements()).containsExactly("BEGIN", "COMMIT", "BEGIN"); + assertThat(client.requests).hasSize(2); + assertThat(client.requests.get(1).baton()).isEqualTo(BATON); + } + + @Test + void rollbackBeginsNextTransactionOnTheSameStream() throws SQLException { + ScriptedClient client = new ScriptedClient((request, index) -> successResponse(request)); + LibsqlConnection connection = new LibsqlConnection(CONFIG, client); + connection.setAutoCommit(false); + + connection.rollback(); + + assertThat(client.executedStatements()).containsExactly("BEGIN", "ROLLBACK", "BEGIN"); + assertThat(client.requests).hasSize(2); + assertThat(client.requests.get(1).baton()).isEqualTo(BATON); + } + + @Test + void enablingAutoCommitCommitsAndClosesTheOpenTransaction() throws SQLException { + ScriptedClient client = new ScriptedClient((request, index) -> successResponse(request)); + LibsqlConnection connection = new LibsqlConnection(CONFIG, client); + connection.setAutoCommit(false); + + connection.setAutoCommit(true); + + assertThat(client.executedStatements()).containsExactly("BEGIN", "COMMIT"); + assertThat(client.requests.get(1).baton()).isEqualTo(BATON); + } + @ParameterizedTest @ValueSource(strings = {"COMMIT", "ROLLBACK"}) void serverErrorIsPropagatedAndTransactionStateIsKept(String command) throws SQLException { @@ -97,6 +181,19 @@ private static void endTransaction(LibsqlConnection connection, String command) } } + private static PipelineRespBody successResponse(PipelineReqBody request) { + List results = new ArrayList<>(); + for (StreamRequest streamRequest : request.requests()) { + if (streamRequest instanceof StreamRequest.Close) { + results.add(new StreamResult.Ok(StreamResponse.Close.INSTANCE)); + } else { + results.add(new StreamResult.Ok( + new StreamResponse.Execute(new StmtResult(List.of(), List.of(), 0, null)))); + } + } + return new PipelineRespBody(BATON, null, results); + } + private static String executedSql(PipelineReqBody request) { if (request.requests().isEmpty() || !(request.requests().get(0) instanceof StreamRequest.Execute execute)) { @@ -105,13 +202,6 @@ private static String executedSql(PipelineReqBody request) { 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 @@ -132,5 +222,14 @@ public PipelineRespBody sendPipeline(PipelineReqBody request, URI targetUri) { requests.add(request); return handler.handle(request, requests.size() - 1); } + + private List executedStatements() { + return requests.stream() + .flatMap(request -> request.requests().stream()) + .filter(StreamRequest.Execute.class::isInstance) + .map(StreamRequest.Execute.class::cast) + .map(execute -> execute.stmt().sql()) + .toList(); + } } }