Skip to content

fix(jdbc): serialize LibsqlConnection state under one lock #153

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
magnusp merged 1 commit into from
Oct 23, 2026
Merged

fix(jdbc): serialize LibsqlConnection state under one lock #153

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
248 changes: 154 additions & 94 deletions libsql-jdbc/src/main/java/com/github/magnusp/libsql/jdbc/LibsqlConnection.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
*
* <p>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.
*
* <p><strong>Lock ordering.</strong> 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;
Expand Down Expand Up @@ -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);
}
}

Expand All @@ -145,36 +169,38 @@ public int[] executeBatch(List<Stmt> statements) throws SQLException {
* driver needs no follow-up query and no extra round trip.
*/
List<StmtResult> executeBatchResults(List<Stmt> statements) throws SQLException {
checkClosed();
if (statements == null || statements.isEmpty()) {
return List.of();
}

try {
List<BatchStep> 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<BatchStep> 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);
}
}

Expand Down Expand Up @@ -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.
*
* <p>Callers hold {@link #stateLock}.
*/
private boolean beginPending() {
return !autoCommit && baton == null;
}

/** Callers hold {@link #stateLock}. */
private void beginTransaction() throws SQLException {
PipelineRespBody resp;
try {
Expand All @@ -267,13 +296,15 @@ 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()) {
this.baseUrl = URI.create(resp.baseUrl());
}
}

/** Callers hold {@link #stateLock}. */
private void closeStreamSilently() {
if (baton != null) {
try {
Expand Down Expand Up @@ -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.
Expand Down Expand Up @@ -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;
}
}

/**
Expand All @@ -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
Expand All @@ -455,32 +505,40 @@ 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 {
setTransactionMode(TransactionMode.from(mode));
}

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
Expand Down Expand Up @@ -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;
Expand Down
Loading
Loading