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 06467e6..879d6b1 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 @@ -40,11 +40,31 @@ import java.util.Properties; import java.util.concurrent.Executor; +/** + * A JDBC connection whose state is serialized on one connection-level lock, the way sqlite-jdbc + * serializes every call on its {@code DB} object. + * + *

Hrana over HTTP requires serialized requests per stream and consumes a baton once, so each + * pipeline round trip and each transaction transition ({@link #setAutoCommit(boolean)}, {@link + * #commit()}, {@link #rollback()}, {@link #close()}) runs while holding {@link #stateLock}. + * Without it, two statements on one non-auto-commit connection can read the same baton, a commit + * can land between another thread's begin and its send, and a close can leave a stream open. + * + *

Lock ordering. A statement's {@code stateLock} is never held across a call + * into the connection, so the only possible order is statement lock, then connection lock. {@code + * LibsqlStatement.checkClosed()} releases its statement lock before calling {@link + * #checkClosed()}, and both execute paths call {@link #executeStatement(Stmt)} and {@link + * #executeBatchResults(List)} after their statement critical sections have exited. The connection + * never calls back into a statement, so the order cannot invert and the two locks cannot deadlock. + */ public class LibsqlConnection implements Connection { private final LibsqlClientConfig config; private final LibsqlHttpClient client; + /** Guards every field below and every pipeline round trip. See the class javadoc. */ + private final Object stateLock = new Object(); + private boolean autoCommit = true; private boolean readOnly = false; private int transactionIsolation = TRANSACTION_SERIALIZABLE; @@ -94,39 +114,43 @@ public LibsqlClientConfig getConfig() { } public void checkClosed() throws SQLException { - if (closed) { - throw new SQLException("Connection is closed"); + synchronized (stateLock) { + if (closed) { + throw new SQLException("Connection is closed"); + } } } public StmtResult executeStatement(Stmt stmt) throws SQLException { - checkClosed(); - try { - if (autoCommit) { - // One-shot execution with single-roundtrip close (prevents Bug 1 128-stream 10s stall) - return client.executeOneShot(stmt); - } else { - ensureTransactionStarted(); - PipelineReqBody req = new PipelineReqBody(baton, List.of(new StreamRequest.Execute(stmt))); - PipelineRespBody resp = client.sendPipeline(req, baseUrl); - updateStreamState(resp); - - if (resp.results().isEmpty()) { - throw new SQLException("Empty result received from server"); - } - StreamResult sr = resp.results().get(0); - return switch (sr) { - case StreamResult.Ok ok -> { - if (ok.response() instanceof StreamResponse.Execute exec) { - yield exec.result(); - } - throw new SQLException("Unexpected response: " + ok.response()); + synchronized (stateLock) { + checkClosed(); + try { + if (autoCommit) { + // One-shot execution with single-roundtrip close (prevents Bug 1 128-stream 10s stall) + return client.executeOneShot(stmt); + } else { + ensureTransactionStarted(); + PipelineReqBody req = new PipelineReqBody(baton, List.of(new StreamRequest.Execute(stmt))); + PipelineRespBody resp = client.sendPipeline(req, baseUrl); + updateStreamState(resp); + + if (resp.results().isEmpty()) { + throw new SQLException("Empty result received from server"); } - case StreamResult.Error err -> throw LibsqlErrorMapper.toSQLException(err.error()); - }; + StreamResult sr = resp.results().get(0); + return switch (sr) { + case StreamResult.Ok ok -> { + if (ok.response() instanceof StreamResponse.Execute exec) { + yield exec.result(); + } + throw new SQLException("Unexpected response: " + ok.response()); + } + case StreamResult.Error err -> throw LibsqlErrorMapper.toSQLException(err.error()); + }; + } + } catch (LibsqlException e) { + throw LibsqlErrorMapper.toSQLException(e); } - } catch (LibsqlException e) { - throw LibsqlErrorMapper.toSQLException(e); } } @@ -145,36 +169,38 @@ public int[] executeBatch(List statements) throws SQLException { * driver needs no follow-up query and no extra round trip. */ List executeBatchResults(List statements) throws SQLException { - checkClosed(); - if (statements == null || statements.isEmpty()) { - return List.of(); - } - - try { - List steps = new ArrayList<>(statements.size()); - for (int i = 0; i < statements.size(); i++) { - BatchCond condition = i == 0 ? null : new BatchCond.StepOk(i - 1); - steps.add(new BatchStep(condition, statements.get(i))); + synchronized (stateLock) { + checkClosed(); + if (statements == null || statements.isEmpty()) { + return List.of(); } - Batch batch = new Batch(steps); - if (autoCommit) { - // One-shot batch - PipelineReqBody req = new PipelineReqBody(null, List.of( - new StreamRequest.BatchReq(batch), - StreamRequest.Close.INSTANCE - )); - PipelineRespBody resp = client.sendPipeline(req, baseUrl); - return extractBatchResults(resp, statements.size()); - } else { - ensureTransactionStarted(); - PipelineReqBody req = new PipelineReqBody(baton, List.of(new StreamRequest.BatchReq(batch))); - PipelineRespBody resp = client.sendPipeline(req, baseUrl); - updateStreamState(resp); - return extractBatchResults(resp, statements.size()); + try { + List steps = new ArrayList<>(statements.size()); + for (int i = 0; i < statements.size(); i++) { + BatchCond condition = i == 0 ? null : new BatchCond.StepOk(i - 1); + steps.add(new BatchStep(condition, statements.get(i))); + } + Batch batch = new Batch(steps); + + if (autoCommit) { + // One-shot batch + PipelineReqBody req = new PipelineReqBody(null, List.of( + new StreamRequest.BatchReq(batch), + StreamRequest.Close.INSTANCE + )); + PipelineRespBody resp = client.sendPipeline(req, baseUrl); + return extractBatchResults(resp, statements.size()); + } else { + ensureTransactionStarted(); + PipelineReqBody req = new PipelineReqBody(baton, List.of(new StreamRequest.BatchReq(batch))); + PipelineRespBody resp = client.sendPipeline(req, baseUrl); + updateStreamState(resp); + return extractBatchResults(resp, statements.size()); + } + } catch (LibsqlException e) { + throw LibsqlErrorMapper.toSQLException(e); } - } catch (LibsqlException e) { - throw LibsqlErrorMapper.toSQLException(e); } } @@ -236,11 +262,14 @@ private void ensureTransactionStarted() throws SQLException { * The begin-pending state is derived rather than stored: auto-commit is off and no stream is * open, so the next statement must begin the transaction. A separate boolean would be a second * source of truth for a fact the baton already carries. + * + *

Callers hold {@link #stateLock}. */ private boolean beginPending() { return !autoCommit && baton == null; } + /** Callers hold {@link #stateLock}. */ private void beginTransaction() throws SQLException { PipelineRespBody resp; try { @@ -267,6 +296,7 @@ private void beginTransaction() throws SQLException { } } + /** Callers hold {@link #stateLock}. */ private void updateStreamState(PipelineRespBody resp) { this.baton = resp.baton(); if (resp.baseUrl() != null && !resp.baseUrl().isBlank()) { @@ -274,6 +304,7 @@ private void updateStreamState(PipelineRespBody resp) { } } + /** Callers hold {@link #stateLock}. */ private void closeStreamSilently() { if (baton != null) { try { @@ -334,40 +365,49 @@ public String nativeSQL(String sql) throws SQLException { */ @Override public void setAutoCommit(boolean autoCommit) throws SQLException { - checkClosed(); - if (this.autoCommit == autoCommit) return; - if (autoCommit) { - endTransaction("COMMIT"); - this.autoCommit = true; - } else { - this.autoCommit = false; + synchronized (stateLock) { + checkClosed(); + if (this.autoCommit == autoCommit) return; + if (autoCommit) { + endTransaction("COMMIT"); + this.autoCommit = true; + } else { + this.autoCommit = false; + } } } @Override public boolean getAutoCommit() throws SQLException { - checkClosed(); - return autoCommit; + synchronized (stateLock) { + checkClosed(); + return autoCommit; + } } @Override public void commit() throws SQLException { - checkClosed(); - if (autoCommit) { - throw new SQLException("Cannot commit when in autoCommit mode"); + synchronized (stateLock) { + checkClosed(); + if (autoCommit) { + throw new SQLException("Cannot commit when in autoCommit mode"); + } + endTransaction("COMMIT"); } - endTransaction("COMMIT"); } @Override public void rollback() throws SQLException { - checkClosed(); - if (autoCommit) { - throw new SQLException("Cannot rollback when in autoCommit mode"); + synchronized (stateLock) { + checkClosed(); + if (autoCommit) { + throw new SQLException("Cannot rollback when in autoCommit mode"); + } + endTransaction("ROLLBACK"); } - endTransaction("ROLLBACK"); } + /** Callers hold {@link #stateLock}. */ private void endTransaction(String sql) throws SQLException { if (beginPending()) { // No transaction has started, so there is nothing to send and nothing to close. @@ -404,15 +444,19 @@ private void endTransaction(String sql) throws SQLException { @Override public void close() throws SQLException { - if (!closed) { - closed = true; - closeStreamSilently(); + synchronized (stateLock) { + if (!closed) { + closed = true; + closeStreamSilently(); + } } } @Override public boolean isClosed() throws SQLException { - return closed; + synchronized (stateLock) { + return closed; + } } /** @@ -423,23 +467,29 @@ public boolean isClosed() throws SQLException { */ @Override public DatabaseMetaData getMetaData() throws SQLException { - checkClosed(); - if (meta == null) { - meta = new LibsqlDatabaseMetaData(this); + synchronized (stateLock) { + checkClosed(); + if (meta == null) { + meta = new LibsqlDatabaseMetaData(this); + } + return meta; } - return meta; } @Override public void setReadOnly(boolean readOnly) throws SQLException { - checkClosed(); - this.readOnly = readOnly; + synchronized (stateLock) { + checkClosed(); + this.readOnly = readOnly; + } } @Override public boolean isReadOnly() throws SQLException { - checkClosed(); - return readOnly; + synchronized (stateLock) { + checkClosed(); + return readOnly; + } } @Override @@ -455,11 +505,13 @@ public String getCatalog() throws SQLException { * 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"); + synchronized (stateLock) { + checkClosed(); + if (mode == null) { + throw new IllegalArgumentException("Transaction mode cannot be null"); + } + this.transactionMode = mode; } - this.transactionMode = mode; } public void setTransactionMode(String mode) throws SQLException { @@ -467,20 +519,26 @@ public void setTransactionMode(String mode) throws SQLException { } public TransactionMode getTransactionMode() throws SQLException { - checkClosed(); - return transactionMode; + synchronized (stateLock) { + checkClosed(); + return transactionMode; + } } @Override public void setTransactionIsolation(int level) throws SQLException { - checkClosed(); - this.transactionIsolation = level; + synchronized (stateLock) { + checkClosed(); + this.transactionIsolation = level; + } } @Override public int getTransactionIsolation() throws SQLException { - checkClosed(); - return transactionIsolation; + synchronized (stateLock) { + checkClosed(); + return transactionIsolation; + } } @Override @@ -601,7 +659,9 @@ public PreparedStatement prepareStatement(String sql, String[] columnNames) thro @Override public boolean isValid(int timeout) throws SQLException { - if (closed) return false; + synchronized (stateLock) { + if (closed) return false; + } try { executeStatement(new Stmt("SELECT 1", null, List.of(), List.of(), false)); return true; diff --git a/libsql-jdbc/src/test/java/com/github/magnusp/libsql/jdbc/LibsqlStatementConcurrencyTest.java b/libsql-jdbc/src/test/java/com/github/magnusp/libsql/jdbc/LibsqlStatementConcurrencyTest.java index bf315cc..08dc22b 100644 --- a/libsql-jdbc/src/test/java/com/github/magnusp/libsql/jdbc/LibsqlStatementConcurrencyTest.java +++ b/libsql-jdbc/src/test/java/com/github/magnusp/libsql/jdbc/LibsqlStatementConcurrencyTest.java @@ -1,20 +1,34 @@ package com.github.magnusp.libsql.jdbc; +import com.github.magnusp.libsql.client.LibsqlClientConfig; +import com.github.magnusp.libsql.client.LibsqlHttpClient; import com.github.magnusp.libsql.hrana.codec.Col; +import com.github.magnusp.libsql.hrana.codec.PipelineReqBody; +import com.github.magnusp.libsql.hrana.codec.PipelineRespBody; import com.github.magnusp.libsql.hrana.codec.Row; 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 com.github.magnusp.libsql.hrana.codec.Value; import org.junit.jupiter.api.Test; +import java.net.URI; import java.sql.ResultSet; import java.sql.SQLException; +import java.util.ArrayList; +import java.util.HashSet; import java.util.List; +import java.util.Set; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.CyclicBarrier; +import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; import static org.assertj.core.api.Assertions.assertThat; @@ -26,6 +40,8 @@ class LibsqlStatementConcurrencyTest { List.of(new Row(List.of(Value.of(1L)))), 0, null); + private static final LibsqlClientConfig CONFIG = + new LibsqlClientConfig("http://localhost:1", null); @Test void plainExecuteQueryDoesNotAttachAnOpenResultSetWhenCloseWinsTheRace() throws Exception { @@ -85,6 +101,129 @@ void executeQueryOnAClosedStatementReturnsAClosedResultSet() throws SQLException assertThat(resultSet.isClosed()).isTrue(); } + /** + * Two threads share one non-auto-commit connection. Hrana consumes a baton once and allows one + * request per stream at a time, so the two lazily started transactions must not both send + * {@code BEGIN} from the same empty stream state. + */ + @Test + void twoThreadsOnOneNonAutoCommitConnectionDoNotReuseABaton() throws Exception { + ProtocolClient client = new ProtocolClient(new CyclicBarrier(2)); + LibsqlConnection connection = new LibsqlConnection(CONFIG, client); + connection.setAutoCommit(false); + LibsqlStatement first = new LibsqlStatement(connection); + LibsqlStatement second = new LibsqlStatement(connection); + ExecutorService executor = Executors.newFixedThreadPool(2); + try { + Future firstWork = executor.submit(() -> executeRepeatedly(first, 20)); + Future secondWork = executor.submit(() -> executeRepeatedly(second, 20)); + + assertThat(failureOf(firstWork, secondWork)).isNull(); + assertThat(client.violations()).as("protocol violations").isEmpty(); + } finally { + executor.shutdownNow(); + } + } + + /** + * A commit from another thread must not land between the lazy transaction begin and the + * statement sent on the new baton. + */ + @Test + void commitDoesNotInterleaveWithTheLazyTransactionBegin() throws Exception { + ProtocolClient client = new ProtocolClient(null); + LibsqlConnection connection = new LibsqlConnection(CONFIG, client); + connection.setAutoCommit(false); + LibsqlStatement statement = new LibsqlStatement(connection); + ExecutorService executor = Executors.newSingleThreadExecutor(); + try { + Future work = executor.submit(() -> statement.execute("SELECT 1")); + assertThat(client.awaitStatementInFlight()).isTrue(); + + AtomicReference commitFailure = new AtomicReference<>(); + joinAndCheck(commitInThread(connection, commitFailure), commitFailure); + + Throwable workFailure = failureOf(work); + assertThat(client.violations()).as("protocol violations").isEmpty(); + assertThat(workFailure).isNull(); + } finally { + executor.shutdownNow(); + } + } + + /** + * A close from another thread must not race the statement's round trip, and it must leave no + * server-side stream open. + */ + @Test + void closeDoesNotInterleaveWithExecuteAndLeavesNoStreamOpen() throws Exception { + ProtocolClient client = new ProtocolClient(null); + LibsqlConnection connection = new LibsqlConnection(CONFIG, client); + connection.setAutoCommit(false); + LibsqlStatement statement = new LibsqlStatement(connection); + ExecutorService executor = Executors.newSingleThreadExecutor(); + try { + Future work = executor.submit(() -> statement.execute("SELECT 1")); + assertThat(client.awaitStatementInFlight()).isTrue(); + + AtomicReference closeFailure = new AtomicReference<>(); + joinAndCheck(closeInThread(connection, closeFailure), closeFailure); + + Throwable workFailure = failureOf(work); + assertThat(client.violations()).as("protocol violations").isEmpty(); + assertThat(workFailure).isNull(); + assertThat(client.hasOpenStream()).isFalse(); + assertThat(connection.isClosed()).isTrue(); + } finally { + executor.shutdownNow(); + } + } + + private static Void executeRepeatedly(LibsqlStatement statement, int count) throws SQLException { + for (int i = 0; i < count; i++) { + statement.execute("SELECT " + i); + } + return null; + } + + private static Thread commitInThread( + LibsqlConnection connection, AtomicReference failure) { + return runInThread(() -> connection.commit(), failure); + } + + private static Thread closeInThread( + LibsqlConnection connection, AtomicReference failure) { + return runInThread(() -> connection.close(), failure); + } + + private static Thread runInThread(ThrowingRunnable action, AtomicReference failure) { + Thread thread = new Thread(() -> { + try { + action.run(); + } catch (Throwable t) { + failure.set(t); + } + }); + thread.start(); + return thread; + } + + private static Throwable failureOf(Future... futures) throws Exception { + for (Future future : futures) { + try { + future.get(10, TimeUnit.SECONDS); + } catch (ExecutionException e) { + return e.getCause(); + } + } + return null; + } + + @FunctionalInterface + private interface ThrowingRunnable { + void run() throws Exception; + } + private static Thread closeInThread( LibsqlStatement statement, AtomicReference failure) { Thread thread = new Thread(() -> { @@ -159,6 +298,144 @@ protected void beforeResultSetAssignment() throws SQLException { } } + /** + * Models the baton rules the connection must uphold: one open stream per connection, one + * request on a stream at a time, and a baton that is consumed by its request. Every rule break + * is recorded instead of thrown, so a failing case reports why the interleaving was invalid. + */ + private static final class ProtocolClient extends LibsqlHttpClient { + + private final Object lock = new Object(); + private final List violations = new ArrayList<>(); + private final Set inFlightBatons = new HashSet<>(); + private final CountDownLatch statementInFlight = new CountDownLatch(1); + private final AtomicBoolean statementWindowUsed = new AtomicBoolean(); + private final CyclicBarrier beginRace; + private int inFlightRequests; + private String liveBaton; + private int batonCount; + private boolean streamOpen; + + private ProtocolClient(CyclicBarrier beginRace) { + super(CONFIG); + this.beginRace = beginRace; + } + + @Override + public PipelineRespBody sendPipeline(PipelineReqBody request, URI targetUri) { + String baton = request.baton(); + StreamRequest first = request.requests().get(0); + String sql = first instanceof StreamRequest.Execute execute + ? execute.stmt().sql() + : null; + boolean close = first instanceof StreamRequest.Close; + + synchronized (lock) { + if (inFlightRequests > 0) { + violations.add("concurrent round trip while " + inFlightRequests + + " was in flight: " + sql); + } + if (baton != null && inFlightBatons.contains(baton)) { + violations.add("baton reused while in flight: " + baton); + } else if (baton != null && (!streamOpen || !baton.equals(liveBaton))) { + violations.add("request on a stream that is not open: " + baton); + } else if (baton == null && streamOpen) { + violations.add("new stream while another is open"); + } + inFlightRequests++; + if (baton != null) { + inFlightBatons.add(baton); + } + } + try { + if (isBegin(sql)) { + awaitBeginRace(); + } else if (sql != null && !isTransactionEnd(sql)) { + if (statementInFlight.getCount() > 0) { + statementInFlight.countDown(); + if (statementWindowUsed.compareAndSet(false, true)) { + pause(); + } + } + } + synchronized (lock) { + if (close) { + liveBaton = null; + streamOpen = false; + } else { + liveBaton = "baton-" + (++batonCount); + streamOpen = true; + } + return new PipelineRespBody( + close ? null : liveBaton, null, responsesFor(request)); + } + } finally { + synchronized (lock) { + inFlightRequests--; + if (baton != null) { + inFlightBatons.remove(baton); + } + } + } + } + + private void awaitBeginRace() { + if (beginRace == null) { + return; + } + try { + beginRace.await(200, TimeUnit.MILLISECONDS); + } catch (Exception expected) { + // The fixed connection holds its lock, so the second begin never arrives. + } + } + + private static void pause() { + try { + Thread.sleep(50); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + + private static boolean isBegin(String sql) { + return sql != null && sql.startsWith("BEGIN"); + } + + private static boolean isTransactionEnd(String sql) { + return "COMMIT".equals(sql) || "ROLLBACK".equals(sql); + } + + private static List responsesFor(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 results; + } + + private boolean awaitStatementInFlight() throws InterruptedException { + return statementInFlight.await(5, TimeUnit.SECONDS); + } + + private List violations() { + synchronized (lock) { + return List.copyOf(violations); + } + } + + private boolean hasOpenStream() { + synchronized (lock) { + return streamOpen; + } + } + } + private static final class RecordingConnection extends LibsqlConnection { private final StmtResult result;