diff --git a/libsql-jdbc/src/main/java/com/github/magnusp/libsql/jdbc/LibsqlPreparedStatement.java b/libsql-jdbc/src/main/java/com/github/magnusp/libsql/jdbc/LibsqlPreparedStatement.java
index 049ff6e..d21452c 100644
--- a/libsql-jdbc/src/main/java/com/github/magnusp/libsql/jdbc/LibsqlPreparedStatement.java
+++ b/libsql-jdbc/src/main/java/com/github/magnusp/libsql/jdbc/LibsqlPreparedStatement.java
@@ -64,9 +64,9 @@ public ResultSet executeQuery() throws SQLException {
if (result.cols() == null || result.cols().isEmpty()) {
throw new SQLException("Query does not return results");
}
- this.currentResultSet = new LibsqlResultSet(this, result, sql, false);
- this.updateCount = -1;
- return currentResultSet;
+ LibsqlResultSet resultSet = new LibsqlResultSet(this, result, sql, false);
+ assignResultSet(resultSet, false);
+ return resultSet;
}
@Override
@@ -217,17 +217,7 @@ private record Token(String word, int depth) {}
@Override
public boolean execute() throws SQLException {
- StmtResult result = executeCurrent();
- updateGeneratedKeys(sql, result);
- if (result.cols() != null && !result.cols().isEmpty()) {
- this.currentResultSet = new LibsqlResultSet(this, result, sql, false);
- this.updateCount = -1;
- return true;
- } else {
- this.currentResultSet = null;
- this.updateCount = result.affectedRowCount();
- return false;
- }
+ return executeResult(sql, executeCurrent());
}
private StmtResult executeCurrent() throws SQLException {
diff --git a/libsql-jdbc/src/main/java/com/github/magnusp/libsql/jdbc/LibsqlStatement.java b/libsql-jdbc/src/main/java/com/github/magnusp/libsql/jdbc/LibsqlStatement.java
index 4a5e2b5..973a68c 100644
--- a/libsql-jdbc/src/main/java/com/github/magnusp/libsql/jdbc/LibsqlStatement.java
+++ b/libsql-jdbc/src/main/java/com/github/magnusp/libsql/jdbc/LibsqlStatement.java
@@ -37,12 +37,17 @@
*
Matching xerial's execute-after-close behavior, {@link #execute(String)}, {@link
* #executeQuery(String)}, {@link #executeUpdate(String)}, and {@link #getConnection()} keep
* working after {@link #close()}. This deliberately copies a likely xerial bug for drop-in
- * compatibility; see the issue #47 contract decision.
+ * compatibility; see the issue #47 contract decision. A result set produced after close is
+ * returned closed, so a closed statement never holds an open result set.
+ * {@link #close()} and the execute methods are safe when called concurrently. State
+ * transitions share one lock; when close wins a race, the execute closes and discards its
+ * result set instead of attaching it to the closed statement.
*
*/
public class LibsqlStatement implements Statement {
protected final LibsqlConnection connection;
+ private final Object stateLock = new Object();
protected LibsqlResultSet currentResultSet = null;
protected long updateCount = -1;
protected boolean closed = false;
@@ -56,9 +61,17 @@ public LibsqlStatement(LibsqlConnection connection) {
this.connection = connection;
}
+ /**
+ * Test seam. Called after the closed check and before a result set is attached to the
+ * statement, so a test can force a close to interleave with execute.
+ */
+ protected void beforeResultSetAssignment() throws SQLException {}
+
protected void checkClosed() throws SQLException {
- if (closed) {
- throw new SQLException("Statement is closed");
+ synchronized (stateLock) {
+ if (closed) {
+ throw new SQLException("Statement is closed");
+ }
}
if (connection != null) {
connection.checkClosed();
@@ -71,10 +84,28 @@ public ResultSet executeQuery(String sql) throws SQLException {
if (result.cols() == null || result.cols().isEmpty()) {
throw new SQLException("Query did not produce a ResultSet: " + sql);
}
- this.currentResultSet = new LibsqlResultSet(this, result, sql, false);
- this.updateCount = -1;
- this.resultSetRequested = true;
- return currentResultSet;
+ LibsqlResultSet resultSet = new LibsqlResultSet(this, result, sql, false);
+ assignResultSet(resultSet, true);
+ return resultSet;
+ }
+
+ protected void assignResultSet(LibsqlResultSet resultSet, boolean requested)
+ throws SQLException {
+ if (!isClosed()) {
+ beforeResultSetAssignment();
+ }
+ synchronized (stateLock) {
+ if (closed) {
+ closeResultSet(resultSet);
+ currentResultSet = null;
+ resultSetRequested = false;
+ updateCount = -1;
+ return;
+ }
+ currentResultSet = resultSet;
+ updateCount = -1;
+ resultSetRequested = requested;
+ }
}
@Override
@@ -85,18 +116,34 @@ public int executeUpdate(String sql) throws SQLException {
@Override
public boolean execute(String sql) throws SQLException {
- StmtResult result = executeSql(sql);
- updateGeneratedKeys(sql, result);
- if (result.cols() != null && !result.cols().isEmpty()) {
- this.currentResultSet = new LibsqlResultSet(this, result, sql, false);
- this.updateCount = -1;
- this.resultSetRequested = false;
- return true;
- } else {
- this.currentResultSet = null;
- this.updateCount = result.affectedRowCount();
- this.resultSetRequested = false;
- return false;
+ return executeResult(sql, executeSql(sql));
+ }
+
+ protected boolean executeResult(String sql, StmtResult result) throws SQLException {
+ boolean hasResultSet = result.cols() != null && !result.cols().isEmpty();
+ LibsqlResultSet resultSet =
+ hasResultSet ? new LibsqlResultSet(this, result, sql, false) : null;
+
+ if (hasResultSet && !isClosed()) {
+ beforeResultSetAssignment();
+ }
+
+ synchronized (stateLock) {
+ if (closed) {
+ clearGeneratedKeys();
+ if (resultSet != null) {
+ closeResultSet(resultSet);
+ }
+ currentResultSet = null;
+ resultSetRequested = false;
+ updateCount = hasResultSet ? -1 : result.affectedRowCount();
+ return hasResultSet;
+ }
+ updateGeneratedKeys(sql, result);
+ currentResultSet = resultSet;
+ updateCount = hasResultSet ? -1 : result.affectedRowCount();
+ resultSetRequested = false;
+ return hasResultSet;
}
}
@@ -106,44 +153,58 @@ private StmtResult executeSql(String sql) throws SQLException {
}
protected void closeCurrentResultSet() throws SQLException {
- LibsqlResultSet resultSet = currentResultSet;
- currentResultSet = null;
- resultSetRequested = false;
- if (resultSet != null) {
- closeResultSet(resultSet);
+ synchronized (stateLock) {
+ LibsqlResultSet resultSet = currentResultSet;
+ currentResultSet = null;
+ resultSetRequested = false;
+ if (resultSet != null) {
+ closeResultSet(resultSet);
+ }
}
}
private void closeResultSet(LibsqlResultSet resultSet) throws SQLException {
- boolean wasClosed = closed;
- boolean wasClosing = closing;
- closed = false;
- closing = true;
- try {
- resultSet.close();
- } finally {
- closed = wasClosed;
- closing = wasClosing;
+ synchronized (stateLock) {
+ boolean wasClosed = closed;
+ boolean wasClosing = closing;
+ closed = false;
+ closing = true;
+ try {
+ resultSet.close();
+ } finally {
+ closed = wasClosed;
+ closing = wasClosing;
+ }
}
}
protected void updateGeneratedKeys(String sql, StmtResult result) throws SQLException {
- clearGeneratedKeys();
- if (sql != null && result != null && result.lastInsertRowid() != null && isInsertSql(sql)) {
- long rowid = result.lastInsertRowid();
- StmtResult keys = new StmtResult(
- List.of(new Col("last_insert_rowid()", null)),
- List.of(new Row(List.of(Value.of(rowid)))),
- 0,
- rowid);
- generatedKeys = new LibsqlResultSet(this, keys);
+ synchronized (stateLock) {
+ clearGeneratedKeys();
+ if (sql != null
+ && result != null
+ && result.lastInsertRowid() != null
+ && isInsertSql(sql)) {
+ long rowid = result.lastInsertRowid();
+ StmtResult keys = new StmtResult(
+ List.of(new Col("last_insert_rowid()", null)),
+ List.of(new Row(List.of(Value.of(rowid)))),
+ 0,
+ rowid);
+ generatedKeys = new LibsqlResultSet(this, keys);
+ }
}
}
protected void clearGeneratedKeys() throws SQLException {
- if (generatedKeys != null) {
- generatedKeys.close();
+ synchronized (stateLock) {
+ ResultSet keys = generatedKeys;
generatedKeys = null;
+ if (keys instanceof LibsqlResultSet libsqlKeys) {
+ closeResultSet(libsqlKeys);
+ } else if (keys != null) {
+ keys.close();
+ }
}
}
@@ -244,35 +305,43 @@ private ResultSet emptyGeneratedKeys() {
@Override
public ResultSet getResultSet() throws SQLException {
- checkClosed();
- if (currentResultSet == null) {
- return null;
- }
- if (resultSetRequested) {
- throw new SQLException("ResultSet already requested");
+ synchronized (stateLock) {
+ checkClosed();
+ if (currentResultSet == null) {
+ return null;
+ }
+ if (resultSetRequested) {
+ throw new SQLException("ResultSet already requested");
+ }
+ resultSetRequested = true;
+ return currentResultSet;
}
- resultSetRequested = true;
- return currentResultSet;
}
@Override
public int getUpdateCount() throws SQLException {
- checkClosed();
- return (int) updateCount;
+ synchronized (stateLock) {
+ checkClosed();
+ return (int) updateCount;
+ }
}
@Override
public long getLargeUpdateCount() throws SQLException {
- checkClosed();
- return updateCount;
+ synchronized (stateLock) {
+ checkClosed();
+ return updateCount;
+ }
}
@Override
public boolean getMoreResults() throws SQLException {
- checkClosed();
- closeCurrentResultSet();
- updateCount = -1;
- return false;
+ synchronized (stateLock) {
+ checkClosed();
+ closeCurrentResultSet();
+ updateCount = -1;
+ return false;
+ }
}
@Override
@@ -304,28 +373,31 @@ public Connection getConnection() throws SQLException {
@Override
public void close() throws SQLException {
- if (closing) {
- return;
- }
- closing = true;
- try {
- closed = false;
- clearGeneratedKeys();
- LibsqlResultSet resultSet = currentResultSet;
- currentResultSet = null;
- resultSetRequested = false;
- if (resultSet != null) {
- resultSet.close();
+ synchronized (stateLock) {
+ if (closing) {
+ return;
+ }
+ closing = true;
+ try {
+ LibsqlResultSet resultSet = currentResultSet;
+ currentResultSet = null;
+ resultSetRequested = false;
+ clearGeneratedKeys();
+ if (resultSet != null) {
+ closeResultSet(resultSet);
+ }
+ } finally {
+ closed = true;
+ closing = false;
}
- } finally {
- closed = true;
- closing = false;
}
}
@Override
public boolean isClosed() throws SQLException {
- return closed;
+ synchronized (stateLock) {
+ return closed;
+ }
}
@Override public int getMaxFieldSize() throws SQLException { return 0; }
@@ -359,11 +431,13 @@ public boolean getMoreResults(int current) throws SQLException {
}
@Override
public ResultSet getGeneratedKeys() throws SQLException {
- checkClosed();
- if (generatedKeys == null) {
- generatedKeys = emptyGeneratedKeys();
+ synchronized (stateLock) {
+ checkClosed();
+ if (generatedKeys == null) {
+ generatedKeys = emptyGeneratedKeys();
+ }
+ return generatedKeys;
}
- return generatedKeys;
}
@Override public int executeUpdate(String sql, int autoGeneratedKeys) throws SQLException { return executeUpdate(sql); }
@Override public int executeUpdate(String sql, int[] columnIndexes) throws SQLException { return executeUpdate(sql); }
@@ -376,14 +450,18 @@ public ResultSet getGeneratedKeys() throws SQLException {
@Override public boolean isPoolable() throws SQLException { return false; }
@Override
public void closeOnCompletion() throws SQLException {
- checkClosedForCompletion();
- closeOnCompletion = true;
+ synchronized (stateLock) {
+ checkClosedForCompletion();
+ closeOnCompletion = true;
+ }
}
@Override
public boolean isCloseOnCompletion() throws SQLException {
- checkClosedForCompletion();
- return closeOnCompletion;
+ synchronized (stateLock) {
+ checkClosedForCompletion();
+ return closeOnCompletion;
+ }
}
private void checkClosedForCompletion() throws SQLException {
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
new file mode 100644
index 0000000..bf315cc
--- /dev/null
+++ b/libsql-jdbc/src/test/java/com/github/magnusp/libsql/jdbc/LibsqlStatementConcurrencyTest.java
@@ -0,0 +1,175 @@
+package com.github.magnusp.libsql.jdbc;
+
+import com.github.magnusp.libsql.hrana.codec.Col;
+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.Value;
+import org.junit.jupiter.api.Test;
+
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+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.AtomicReference;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class LibsqlStatementConcurrencyTest {
+
+ private static final StmtResult ROWS = new StmtResult(
+ List.of(new Col("value", null)),
+ List.of(new Row(List.of(Value.of(1L)))),
+ 0,
+ null);
+
+ @Test
+ void plainExecuteQueryDoesNotAttachAnOpenResultSetWhenCloseWinsTheRace() throws Exception {
+ Pause pause = new Pause();
+ PausingStatement statement =
+ new PausingStatement(new RecordingConnection(ROWS), pause);
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ try {
+ Future execute =
+ executor.submit(() -> statement.executeQuery("select value from t"));
+
+ assertThat(pause.awaitReached()).isTrue();
+ AtomicReference closeFailure = new AtomicReference<>();
+ joinAndCheck(closeInThread(statement, closeFailure), closeFailure);
+ pause.resume();
+
+ ResultSet resultSet = execute.get(5, TimeUnit.SECONDS);
+ assertThat(statement.isClosed()).isTrue();
+ assertThat(resultSet.isClosed()).isTrue();
+ } finally {
+ pause.resume();
+ executor.shutdownNow();
+ }
+ }
+
+ @Test
+ void preparedExecuteQueryDoesNotAttachAnOpenResultSetWhenCloseWinsTheRace() throws Exception {
+ Pause pause = new Pause();
+ PausingPreparedStatement statement = new PausingPreparedStatement(
+ new RecordingConnection(ROWS), "select value from t", pause);
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ try {
+ Future execute = executor.submit(() -> statement.executeQuery());
+
+ assertThat(pause.awaitReached()).isTrue();
+ AtomicReference closeFailure = new AtomicReference<>();
+ joinAndCheck(closeInThread(statement, closeFailure), closeFailure);
+ pause.resume();
+
+ ResultSet resultSet = execute.get(5, TimeUnit.SECONDS);
+ assertThat(statement.isClosed()).isTrue();
+ assertThat(resultSet.isClosed()).isTrue();
+ } finally {
+ pause.resume();
+ executor.shutdownNow();
+ }
+ }
+
+ @Test
+ void executeQueryOnAClosedStatementReturnsAClosedResultSet() throws SQLException {
+ LibsqlStatement statement = new LibsqlStatement(new RecordingConnection(ROWS));
+ statement.close();
+
+ ResultSet resultSet = statement.executeQuery("select value from t");
+
+ assertThat(statement.isClosed()).isTrue();
+ assertThat(resultSet.isClosed()).isTrue();
+ }
+
+ private static Thread closeInThread(
+ LibsqlStatement statement, AtomicReference failure) {
+ Thread thread = new Thread(() -> {
+ try {
+ statement.close();
+ } catch (Throwable t) {
+ failure.set(t);
+ }
+ });
+ thread.start();
+ return thread;
+ }
+
+ private static void joinAndCheck(Thread thread, AtomicReference failure)
+ throws InterruptedException {
+ thread.join(5_000);
+ assertThat(thread.isAlive()).isFalse();
+ assertThat(failure.get()).isNull();
+ }
+
+ private static final class Pause {
+ private final CountDownLatch reached = new CountDownLatch(1);
+ private final CountDownLatch resume = new CountDownLatch(1);
+
+ private void await() throws SQLException {
+ reached.countDown();
+ try {
+ if (!resume.await(5, TimeUnit.SECONDS)) {
+ throw new SQLException("Timed out waiting to resume execute");
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new SQLException(e);
+ }
+ }
+
+ private boolean awaitReached() throws InterruptedException {
+ return reached.await(5, TimeUnit.SECONDS);
+ }
+
+ private void resume() {
+ resume.countDown();
+ }
+ }
+
+ private static final class PausingStatement extends LibsqlStatement {
+ private final Pause pause;
+
+ private PausingStatement(LibsqlConnection connection, Pause pause) {
+ super(connection);
+ this.pause = pause;
+ }
+
+ @Override
+ protected void beforeResultSetAssignment() throws SQLException {
+ pause.await();
+ }
+ }
+
+ private static final class PausingPreparedStatement extends LibsqlPreparedStatement {
+ private final Pause pause;
+
+ private PausingPreparedStatement(
+ LibsqlConnection connection, String sql, Pause pause) {
+ super(connection, sql);
+ this.pause = pause;
+ }
+
+ @Override
+ protected void beforeResultSetAssignment() throws SQLException {
+ pause.await();
+ }
+ }
+
+ private static final class RecordingConnection extends LibsqlConnection {
+ private final StmtResult result;
+
+ private RecordingConnection(StmtResult result) {
+ super(null, null);
+ this.result = result;
+ }
+
+ @Override
+ public StmtResult executeStatement(Stmt stmt) {
+ return result;
+ }
+ }
+}