fix(jdbc): keep one connection-scoped Hrana stream for auto-commit

libsql-client/src/main/java/io/github/magnusp/libsql/client/LibsqlClient.java

@@ -11,11 +11,12 @@
  * <p>Both transports expose the same two execution styles. A <em>one-shot</em> call opens a stream,
  * runs the work and closes the stream in as few round trips as possible, so it leaves no server-side
  * state behind and needs no explicit cleanup. An <em>on-stream</em> call reuses a lazily opened
- * interactive stream, which is what a transaction needs: the server keeps the transaction open
- * between calls.
+ * interactive stream. The JDBC driver uses the on-stream style for every statement, so auto-commit
+ * retains SQLite session state such as {@code last_insert_rowid()} and {@code PRAGMA} settings, and
+ * manual transactions stay open between calls.
  *
- * <p>Implementations track their own stream state. {@link #hasOpenStream()} is the single source of
- * truth for whether a transaction begin is still pending.
+ * <p>Implementations track their own stream state. {@link #hasOpenStream()} tells a caller whether
+ * the lazily opened stream still exists, which matters after the server expires an idle stream.
  */
 public interface LibsqlClient extends AutoCloseable {
 

libsql-conformance/README.md

@@ -122,34 +122,33 @@ categories above, these occur in selected tests. They are driver behavior
 rather than edits to a vendored test, so they are described here and are not
 recorded as `# divergence:` lines.
 
-#### Auto-commit holds no stream
-
-An auto-commit statement runs as a one-shot pipeline that executes the
-statement and closes the stream in the same round trip. The driver keeps no
-session stream open between statements, so an idle connection consumes no
-server stream slot and holds no baton for the idle expiry to invalidate.
-Keeping one open only to preserve session state would reintroduce stale-baton
-failures on idle connections.
-
-Because `StmtResult.last_insert_rowid` describes the insert on one stream and
-nothing else, a later `SELECT last_insert_rowid()` runs on a fresh stream and
-returns `0`, even when the same `Statement` inserted a row immediately before.
-xerial returns that insert's rowid instead. Use `Statement.getGeneratedKeys()`
-after the insert, which the driver builds from `StmtResult.last_insert_rowid`
-with no extra round trip, or run the insert and the query inside one explicit
-transaction so both run on the same stream. `LibsqlGeneratedKeysTest` pins the
-generated-keys path.
+#### Auto-commit holds one connection-scoped stream
+
+An auto-commit connection lazily opens one interactive Hrana stream and
+pipelines every statement on it, the same machinery explicit transactions
+use. `last_insert_rowid()`, `PRAGMA` settings, and other session state
+therefore behave as they do on a SQLite connection: they survive between
+statements on the same connection and do not cross connections.
+`Statement.getGeneratedKeys()` still reads `StmtResult.last_insert_rowid`
+without an extra round trip.
+
+sqld expires idle streams (around 10 seconds). When it rejects a baton, the
+driver drops the stream and retries the statement once on a fresh one.
+Auto-commit has no transaction to lose, but session state set before the
+expiry is reset. A pool larger than sqld's stream limit can hit that limit,
+just as idle non-auto-commit connections already can.
 
 #### Transactions begin lazily
 
 `setAutoCommit(false)` sends nothing. The next statement begins the
-transaction on a new stream with `BEGIN`, `BEGIN IMMEDIATE`, or
+transaction on the connection's stream with `BEGIN`, `BEGIN IMMEDIATE`, or
 `BEGIN EXCLUSIVE` according to
 `LibsqlConnection.setTransactionMode(TransactionMode)` (default `DEFERRED`),
 sent as its own pipeline so a failed begin cannot be followed by the
-statement running outside a transaction. `commit()` and `rollback()` close
-the stream and do not begin the next transaction, so an idle non-auto-commit
-connection likewise holds no stream.
+statement running outside a transaction. A connection that used auto-commit
+first keeps that stream, so its session state carries into the transaction.
+`commit()` and `rollback()` close the stream and do not begin the next
+transaction.
 
 The divergence is timing only: an `IMMEDIATE` or `EXCLUSIVE` transaction takes
 its write lock when the first statement begins it, not at

libsql-jdbc/src/main/java/io/github/magnusp/libsql/jdbc/LibsqlConnection.java

@@ -71,6 +71,7 @@ public class LibsqlConnection implements Connection {
     private TransactionMode transactionMode = TransactionMode.DEFERRED;
     private LibsqlDatabaseMetaData meta;
     private int nextSavepointId = 0;
+    private boolean transactionActive = false;
     private boolean transactionStartedBySavepoint = false;
     private final List<String> savepointNames = new ArrayList<>();
 
@@ -102,7 +103,9 @@ public static TransactionMode from(String mode) {
         }
     }
 
-    // Interactive stream state lives in the client; see LibsqlClient.hasOpenStream().
+    // The client owns the Hrana stream. Auto-commit keeps one open between statements so session
+    // state survives; a manual transaction tracks its begin explicitly because the stream may
+    // already be open for auto-commit when the transaction starts.
 
     public LibsqlConnection(LibsqlClientConfig config, LibsqlClient client) {
         this(config, client, null);
@@ -135,11 +138,12 @@ public StmtResult executeStatement(Stmt stmt) throws SQLException {
         synchronized (stateLock) {
             checkClosed();
             if (autoCommit) {
-                // One-shot: single-round-trip close, so no idle stream is left for the server to reap.
-                return sending(() -> client.executeOneShot(stmt));
+                return sendAutoCommit(stmt);
             }
             ensureTransactionStarted();
-            return sendOnStream(stmt);
+            StmtResult result = sendOnStream(stmt);
+            requireTransactionStream();
+            return result;
         }
     }
 
@@ -162,10 +166,11 @@ List<StmtResult> executeBatchResults(List<Stmt> statements) throws SQLException
             Batch batch = new Batch(steps);
             BatchResult result;
             if (autoCommit) {
-                result = sending(() -> client.executeBatchOneShot(batch));
+                result = sendBatchAutoCommit(batch);
             } else {
                 ensureTransactionStarted();
                 result = sendBatchOnStream(batch);
+                requireTransactionStream();
             }
             return extractBatchResults(result, statements.size());
         }
@@ -215,6 +220,62 @@ private <T> T sending(Supplier<T> call) throws SQLException {
         }
     }
 
+    /**
+     * Sends an auto-commit statement on the connection's session stream. sqld expires idle streams
+     * (around 10 seconds); when it rejects the baton, the statement did not run, so the driver drops
+     * the stale stream and retries once on a fresh one. Auto-commit has no transaction to lose.
+     * Session state set before the expiry is gone, by design.
+     *
+     * <p>Callers hold {@link #stateLock}.
+     */
+    private StmtResult sendAutoCommit(Stmt stmt) throws SQLException {
+        try {
+            return client.executeOnStream(stmt);
+        } catch (LibsqlException e) {
+            if (!isExpiredStream(e)) {
+                throw LibsqlErrorMapper.toSQLException(e);
+            }
+            closeStreamQuietly();
+            try {
+                return client.executeOnStream(stmt);
+            } catch (LibsqlException retry) {
+                throw LibsqlErrorMapper.toSQLException(retry);
+            }
+        }
+    }
+
+    /** Auto-commit counterpart of {@link #sendAutoCommit(Stmt)}, with the same one retry. */
+    private BatchResult sendBatchAutoCommit(Batch batch) throws SQLException {
+        try {
+            return client.executeBatchOnStream(batch);
+        } catch (LibsqlException e) {
+            if (!isExpiredStream(e)) {
+                throw LibsqlErrorMapper.toSQLException(e);
+            }
+            closeStreamQuietly();
+            try {
+                return client.executeBatchOnStream(batch);
+            } catch (LibsqlException retry) {
+                throw LibsqlErrorMapper.toSQLException(retry);
+            }
+        }
+    }
+
+    private static boolean isExpiredStream(LibsqlException e) {
+        String code = e.getCode();
+        if (code != null) {
+            String upper = code.toUpperCase(java.util.Locale.ROOT);
+            if (upper.equals("STREAM_EXPIRED")
+                    || upper.equals("BATON_EXPIRED")
+                    || upper.equals("STALE_BATON")) {
+                return true;
+            }
+        }
+        String message = e.getMessage();
+        return message != null
+                && message.toLowerCase(java.util.Locale.ROOT).contains("stream has expired");
+    }
+
     /** Stream sends. Callers hold {@link #stateLock}. */
     private StmtResult sendOnStream(Stmt stmt) throws SQLException {
         return sending(() -> client.executeOnStream(stmt));
@@ -224,27 +285,57 @@ private BatchResult sendBatchOnStream(Batch batch) throws SQLException {
         return sending(() -> client.executeBatchOnStream(batch));
     }
 
-    /** Callers hold {@link #stateLock}. */
+    /**
+     * Fails the transaction when the server closed its stream anyway. Executing the next statement
+     * on a fresh stream would silently run it outside the transaction the caller asked for.
+     * Callers hold {@link #stateLock}.
+     */
+    private void requireTransactionStream() throws SQLException {
+        if (!client.hasOpenStream()) {
+            transactionActive = false;
+            clearSavepoints();
+            throw new SQLException("The server closed the transaction stream");
+        }
+    }
+
+    /**
+     * Begins the manual transaction lazily. An auto-commit stream may already be open when the
+     * first transactional statement arrives, and that stream may have expired while the connection
+     * was idle. The begin did not run in that case, so drop the stale stream and retry once.
+     *
+     * <p>Callers hold {@link #stateLock}.
+     */
     private void ensureTransactionStarted() throws SQLException {
-        if (!beginPending()) return;
+        if (transactionActive) return;
+        Stmt begin = new Stmt(transactionMode.beginSql());
         try {
-            sendOnStream(new Stmt(transactionMode.beginSql()));
+            sendBegin(begin);
+        } catch (LibsqlException e) {
+            if (!isExpiredStream(e)) {
+                closeStreamQuietly();
+                throw LibsqlErrorMapper.toSQLException(e);
+            }
+            closeStreamQuietly();
+            try {
+                sendBegin(begin);
+            } catch (LibsqlException retry) {
+                closeStreamQuietly();
+                throw LibsqlErrorMapper.toSQLException(retry);
+            }
         } catch (SQLException e) {
             // The begin failed, so the fresh stream holds no transaction state.
             closeStreamQuietly();
             throw e;
         }
     }
 
-    /**
-     * The begin-pending state is derived rather than stored: auto-commit is off and the client has
-     * no open stream, so the next statement must begin the transaction. A separate boolean would be
-     * a second source of truth for a fact the client's stream state already carries.
-     *
-     * <p>Callers hold {@link #stateLock}.
-     */
-    private boolean beginPending() {
-        return !autoCommit && !client.hasOpenStream();
+    /** Callers hold {@link #stateLock}. */
+    private void sendBegin(Stmt begin) throws SQLException {
+        client.executeOnStream(begin);
+        if (!client.hasOpenStream()) {
+            throw new SQLException("The server closed the stream before the transaction began");
+        }
+        transactionActive = true;
     }
 
     /** Callers hold {@link #stateLock}. */
@@ -293,13 +384,14 @@ public String nativeSQL(String sql) throws SQLException {
      *
      * <ul>
      *   <li>Disabling auto-commit only records that the next statement must begin a transaction in
-     *       the configured {@link TransactionMode}. No request is sent, so an idle non-auto-commit
-     *       connection holds no Hrana stream and cannot hit sqld's idle stream expiry.
+     *       the configured {@link TransactionMode}. No request is sent. An auto-commit stream that
+     *       was already open stays open and the begin runs on it, so session state set in
+     *       auto-commit survives into the transaction.
      *   <li>{@link #commit()} and {@link #rollback()} end the open transaction and close its stream
      *       without beginning the next one. A commit or rollback with no work sent nothing to the
      *       server, as in xerial.
      *   <li>Re-enabling auto-commit commits the open transaction and ends the Hrana transaction
-     *       stream.
+     *       stream. The next auto-commit statement opens a fresh session stream.
      * </ul>
      *
      * <p>This diverges from xerial only in when the server sees the begin: an {@code IMMEDIATE} or
@@ -351,7 +443,7 @@ public void rollback() throws SQLException {
 
     /** Callers hold {@link #stateLock}. */
     private void endTransaction(String sql) throws SQLException {
-        if (beginPending()) {
+        if (!transactionActive) {
             // No transaction has started, so there is nothing to send and nothing to close.
             return;
         }
@@ -361,6 +453,7 @@ private void endTransaction(String sql) throws SQLException {
         // The server committed or rolled back, so the stream has no state left worth holding.
         // Closing it lets an idle connection drop its server-side stream. On an error above the
         // stream stays, so the caller can retry or roll back on the same stream.
+        transactionActive = false;
         clearSavepoints();
         closeStreamQuietly();
     }
@@ -370,6 +463,7 @@ public void close() throws SQLException {
         synchronized (stateLock) {
             if (!closed) {
                 closed = true;
+                transactionActive = false;
                 clearSavepoints();
                 closeStreamQuietly();
                 if (client != null) {
@@ -563,12 +657,14 @@ private Savepoint setSavepointInternal(String name) throws SQLException {
     private void beginSavepointTransaction(Stmt stmt, String sqlName) throws SQLException {
         try {
             sendOnStream(stmt);
+            requireTransactionStream();
         } catch (SQLException e) {
             // No transaction was started, so the fresh stream holds no state.
             closeStreamQuietly();
             throw e;
         }
         autoCommit = false;
+        transactionActive = true;
         transactionStartedBySavepoint = true;
         savepointNames.clear();
         savepointNames.add(sqlName);
@@ -617,6 +713,7 @@ private void closeTransactionOwnedBySavepoint(String sqlName) {
         savepointNames.subList(index, savepointNames.size()).clear();
         if (index > 0 || !transactionStartedBySavepoint) return;
         transactionStartedBySavepoint = false;
+        transactionActive = false;
         closeStreamQuietly();
     }
 

libsql-jdbc/src/main/java/io/github/magnusp/libsql/jdbc/LibsqlPreparedStatement.java

@@ -68,11 +68,20 @@ public class LibsqlPreparedStatement extends LibsqlStatement implements Prepared
     public LibsqlPreparedStatement(LibsqlConnection connection, String sql) {
         super(connection);
         this.sql = sql;
-        this.returnsResultSet = returnsResultSet(sql);
+        this.returnsResultSet = returnsResultSet(finalStatement(sql));
         this.parameterCount = SqlTokenizer.placeholderCount(sql);
         this.parameterMetaData = new ParameterMetaDataView(this, parameterCount);
     }
 
+    /**
+     * A multi-statement script exposes the final statement's result, so that statement decides
+     * whether the prepared execute returns a result set.
+     */
+    private static String finalStatement(String sql) {
+        List<String> statements = SqlScript.split(sql);
+        return statements.isEmpty() ? sql : statements.get(statements.size() - 1);
+    }
+
     private void setParam(int parameterIndex, Value value) throws SQLException {
         checkClosed();
         int idx = parameterIndex - 1;
@@ -184,7 +193,7 @@ private StmtResult executeCurrent() throws SQLException {
         checkEscapeProcessing(sql);
         checkClosed();
         beginExecution();
-        return connection.executeStatement(buildCurrentStmt());
+        return executePrepared(buildCurrentStmt(), !parameters.isEmpty());
     }
 
     @Override