From e4574743ff553af110a111655056e051fc4c2bec Mon Sep 17 00:00:00 2001 From: Clemens Portele Date: Sat, 29 Aug 2026 13:42:51 +0200 Subject: [PATCH 1/2] fix pooled connections leaking when a CRUD mutation fails at commit CRUD writes (POST/PUT/PATCH/DELETE on items) ran through the rxjava3-jdbc transacted chain, which returns the connection to the pool only when its internal reference counter reaches zero. A COMMIT rejected by the database (deferred constraint, constraint trigger, serialization failure at commit) drives the counter below zero and the lease is never released; enough of these starve the pool and every request fails with "Connection is not available". The CRUD entry points of the SQL feature provider now run through the same JDBC session as the transactions module: one session per mutation, committed on success, rolled back on error, and the connection is returned in a finally block. The rxjava3-jdbc write path (getMutationSource, getMutationFlow, the creator/updater/deletion flows and the transaction guard) is removed. The statements are unchanged, PATCH keeps its delete-and-insert semantics, and a POST with several features is now a single transaction. Errors caused by the submitted data keep being reported as a bad request; a lost connection or an exhausted pool is now reported as a server error instead of "Invalid feature data". The session's connection release is hardened as well: resetting autocommit and closing are isolated so a connection the database has terminated is still given back, a failed COMMIT finalises the session, and the connection is released exactly once. --- .../features/sql/app/FeatureMutationsSql.java | 51 +---- .../sql/domain/FeatureProviderSql.java | 197 ++++++++--------- .../features/sql/domain/SqlClient.java | 19 -- .../features/sql/infra/db/JdbcSqlSession.java | 25 ++- .../infra/db/MutationTransactionGuard.java | 106 --------- .../features/sql/infra/db/SqlClientRx.java | 193 ---------------- .../sql/app/FeatureMutationsSqlSpec.groovy | 8 +- .../sql/infra/db/JdbcSqlSessionSpec.groovy | 47 ++++ .../sql/infra/db/SqlClientRxSpec.groovy | 206 ------------------ 9 files changed, 163 insertions(+), 689 deletions(-) delete mode 100644 xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/infra/db/MutationTransactionGuard.java delete mode 100644 xtraplatform-features-sql/src/test/groovy/de/ii/xtraplatform/features/sql/infra/db/SqlClientRxSpec.groovy diff --git a/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/app/FeatureMutationsSql.java b/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/app/FeatureMutationsSql.java index b42b9f4b4..4c3f67294 100644 --- a/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/app/FeatureMutationsSql.java +++ b/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/app/FeatureMutationsSql.java @@ -11,14 +11,10 @@ import de.ii.xtraplatform.crs.domain.EpsgCrs; import de.ii.xtraplatform.features.domain.SchemaBase.Type; import de.ii.xtraplatform.features.domain.Tuple; -import de.ii.xtraplatform.features.sql.domain.SqlClient; import de.ii.xtraplatform.features.sql.domain.SqlPathDefaults; import de.ii.xtraplatform.features.sql.domain.SqlQueryColumn; import de.ii.xtraplatform.features.sql.domain.SqlQueryMapping; -import de.ii.xtraplatform.features.sql.domain.SqlQueryOptions; import de.ii.xtraplatform.features.sql.domain.SqlQuerySchema; -import de.ii.xtraplatform.streams.domain.Reactive; -import de.ii.xtraplatform.streams.domain.Reactive.Transformer; import java.util.List; import java.util.Objects; import java.util.Optional; @@ -34,60 +30,15 @@ public class FeatureMutationsSql { private static final Logger LOGGER = LoggerFactory.getLogger(FeatureMutationsSql.class); - private final Supplier sqlClient; private final FeatureStoreInsertGenerator generator; private final SqlPathDefaults sqlPathDefaults; public FeatureMutationsSql( - Supplier sqlClient, - FeatureStoreInsertGenerator generator, - SqlPathDefaults sqlPathDefaults) { - this.sqlClient = sqlClient; + FeatureStoreInsertGenerator generator, SqlPathDefaults sqlPathDefaults) { this.generator = generator; this.sqlPathDefaults = sqlPathDefaults; } - public Reactive.Transformer getCreatorFlow( - SqlQueryMapping schema, Object executionContext, Optional id, EpsgCrs crs) { - - RowCursor rowCursor = new RowCursor(schema.getMainTable().getFullPath()); - - String primaryKey = schema.getMainTable().getPrimaryKey(); - - return sqlClient - .get() - .getMutationFlow( - feature -> createInstanceInserts(feature, rowCursor, id, crs, false), - executionContext, - primaryKey, - Optional.empty()); - } - - public Reactive.Transformer getUpdaterFlow( - SqlQueryMapping schema, Object executionContext, String id, EpsgCrs crs) { - - RowCursor rowCursor = new RowCursor(schema.getMainTable().getFullPath()); - - String primaryKey = schema.getMainTable().getPrimaryKey(); - - return sqlClient - .get() - .getMutationFlow( - feature -> createInstanceInserts(feature, rowCursor, Optional.of(id), crs, true), - executionContext, - primaryKey, - Optional.of(id)); - } - - public Reactive.Source getDeletionSource(SqlQueryMapping mapping, String id) { - Supplier>> delete = createInstanceDelete(mapping, id); - - return sqlClient - .get() - .getSourceStream(delete.get().first(), SqlQueryOptions.withColumnTypes(String.class)) - .via(Transformer.map(sqlRow -> (String) sqlRow.getValues().get(0))); - } - List>>> createInstanceInserts( FeatureDataSql feature, RowCursor rowCursor, diff --git a/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/domain/FeatureProviderSql.java b/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/domain/FeatureProviderSql.java index 147c01eb1..dd1a549ee 100644 --- a/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/domain/FeatureProviderSql.java +++ b/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/domain/FeatureProviderSql.java @@ -57,7 +57,6 @@ import de.ii.xtraplatform.features.domain.FeatureTokenDecoder; import de.ii.xtraplatform.features.domain.FeatureTokenSource; import de.ii.xtraplatform.features.domain.FeatureTransactions; -import de.ii.xtraplatform.features.domain.FeatureTransactions.MutationResult.Builder; import de.ii.xtraplatform.features.domain.FeatureTransactions.MutationResult.Type; import de.ii.xtraplatform.features.domain.FilterEncoder; import de.ii.xtraplatform.features.domain.ImmutableDatasetChange; @@ -85,13 +84,10 @@ import de.ii.xtraplatform.features.sql.SqlPathSyntax; import de.ii.xtraplatform.features.sql.app.AggregateStatsQueryGenerator; import de.ii.xtraplatform.features.sql.app.AggregateStatsReaderSql; -import de.ii.xtraplatform.features.sql.app.FeatureDataSql; import de.ii.xtraplatform.features.sql.app.FeatureDecoderSql; -import de.ii.xtraplatform.features.sql.app.FeatureEncoderSql; import de.ii.xtraplatform.features.sql.app.FeatureMutationsSql; import de.ii.xtraplatform.features.sql.app.FeatureQueryEncoderSql; import de.ii.xtraplatform.features.sql.app.FilterEncoderSql; -import de.ii.xtraplatform.features.sql.app.ModifiableFeatureDataSql; import de.ii.xtraplatform.features.sql.app.MutationSchemaDeriver; import de.ii.xtraplatform.features.sql.app.PathParserSql; import de.ii.xtraplatform.features.sql.app.QuerySchemaDeriver; @@ -108,11 +104,7 @@ import de.ii.xtraplatform.services.domain.AuditLog; import de.ii.xtraplatform.services.domain.Scheduler; import de.ii.xtraplatform.streams.domain.Reactive; -import de.ii.xtraplatform.streams.domain.Reactive.RunnableStream; -import de.ii.xtraplatform.streams.domain.Reactive.Sink; -import de.ii.xtraplatform.streams.domain.Reactive.Source; import de.ii.xtraplatform.streams.domain.Reactive.Stream; -import de.ii.xtraplatform.streams.domain.Reactive.Transformer; import de.ii.xtraplatform.values.domain.ValueStore; import java.sql.SQLException; import java.time.ZoneId; @@ -130,10 +122,10 @@ import java.util.concurrent.CompletionException; import java.util.concurrent.CompletionStage; import java.util.function.BiFunction; +import java.util.function.Function; import java.util.function.Predicate; import java.util.stream.Collectors; import java.util.stream.IntStream; -import org.postgresql.util.PSQLException; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.threeten.extra.Interval; @@ -658,7 +650,6 @@ protected boolean onStartup() throws InterruptedException { getData().getNativeTimeZone().orElse(ZoneId.of("UTC"))); this.featureMutationsSql = new FeatureMutationsSql( - this::getSqlClient, new SqlInsertGenerator2( getData().getNativeCrs().orElse(OgcCrs.CRS84), crsTransformerFactory, @@ -1263,8 +1254,9 @@ public MutationResult createFeatures( FeatureTokenSource featureTokenSource, EpsgCrs crs, Optional featureId) { - - return writeFeatures(Type.CREATE, featureType, featureTokenSource, featureId, crs, false); + return runInSession( + Type.CREATE, + session -> session.createFeatures(featureType, featureTokenSource, crs, featureId)); } @Override @@ -1274,13 +1266,9 @@ public MutationResult updateFeature( FeatureTokenSource featureTokenSource, EpsgCrs crs, boolean partial) { - return writeFeatures( + return runInSession( partial ? Type.UPDATE : Type.REPLACE, - featureType, - featureTokenSource, - Optional.of(featureId), - crs, - partial); + session -> session.updateFeature(featureType, featureId, featureTokenSource, crs, partial)); } @Override @@ -1303,27 +1291,7 @@ protected boolean supportsEncryptedProperties() { @Override public MutationResult deleteFeature(String featureType, String id) { - Optional> queryMapping = - Optional.ofNullable(queryMappings.get(featureType)); - - if (queryMapping.isEmpty()) { - throw new IllegalArgumentException( - String.format("Feature type '%s' not found.", featureType)); - } - - Reactive.Source deletionSource = - featureMutationsSql.getDeletionSource(queryMapping.get().get(0), id); - - RunnableStream deletionStream = - deletionSource - .to(Sink.ignore()) - .withResult(ImmutableMutationResult.builder().type(Type.DELETE).hasFeatures(false)) - .handleError(ImmutableMutationResult.Builder::error) - .handleItem(ImmutableMutationResult.Builder::addIds) - .handleEnd(Builder::build) - .on(getStreamRunner()); - - return deletionStream.run().toCompletableFuture().join(); + return runInSession(Type.DELETE, session -> session.deleteFeature(featureType, id)); } @Override @@ -1356,80 +1324,97 @@ public boolean hasGeneratedId(String featureType) { return true; } - private MutationResult writeFeatures( - Type type, - String featureType, - FeatureTokenSource featureTokenSource, - Optional featureId, - EpsgCrs crs, - boolean partial) { - Optional> queryMapping = - Optional.ofNullable(queryMappings.get(featureType)); + /** + * Runs one mutation in a session of its own: committed when the session reports success, rolled + * back otherwise, and the connection is returned to the pool in every case, including a failed + * COMMIT or a connection the database has terminated. + */ + private MutationResult runInSession(Type type, Function mutation) { + Session session; + try { + session = openSession(); + } catch (RuntimeException e) { + // no connection could be leased, so there is nothing to release + return ImmutableMutationResult.builder().type(type).hasFeatures(false).error(e).build(); + } + + try { + MutationResult result = mutation.apply(session); + + if (result.getError().isPresent()) { + session.rollback(); - if (queryMapping.isEmpty()) { - throw new IllegalArgumentException( - String.format("Feature type '%s' not found.", featureType)); + return ImmutableMutationResult.builder() + .from(result) + .error(toMutationError(result.getError().get())) + .build(); + } + + session.commit(); + + return result; + } catch (RuntimeException e) { + session.rollback(); + + return ImmutableMutationResult.builder() + .type(type) + .hasFeatures(false) + .error(toMutationError(e)) + .build(); + } finally { + session.close(); } + } - Transformer featureWriter = - type == Type.CREATE - ? featureMutationsSql.getCreatorFlow(queryMapping.get().get(0), null, featureId, crs) - : featureMutationsSql.getUpdaterFlow( - queryMapping.get().get(0), null, featureId.get(), crs); - - ImmutableMutationResult.Builder builder = - ImmutableMutationResult.builder().type(type).hasFeatures(false); - FeatureTokenStatsCollector statsCollector = new FeatureTokenStatsCollector(builder, crs); - - Source featureSqlSource = - featureTokenSource - .via(statsCollector) - .via( - new FeatureEncoderSql( - queryMapping.get().get(0), - crs, - getNativeCrs(), - crsTransformerFactory, - getData().getNativeTimeZone(), - partial ? Optional.of(FeatureTransactions.PATCH_NULL_VALUE) : Optional.empty(), - getPropertyEncryption())) - .via(Transformer.map(feature -> feature)); - - if (partial) { - featureSqlSource = - featureSqlSource.via( - Transformer.reduce( - ModifiableFeatureDataSql.create(), - (a, b) -> a.getRows().isEmpty() ? b : a.patchWith(b))); + /** + * Errors caused by the submitted data - a statement or a COMMIT rejected by the database, + * unparsable JSON - are reported as a bad request with a generic message; the database message is + * in the debug log and is returned to the client with 'Prefer: handling=strict'. Everything else, + * notably a lost database connection or an exhausted pool, is passed through and reported as a + * server error. + */ + private static Throwable toMutationError(Throwable error) { + for (Throwable t = error; Objects.nonNull(t); t = t.getCause() == t ? null : t.getCause()) { + if (t instanceof JsonParseException) { + return invalidFeatureData(t); + } + if (t instanceof SQLException) { + return isDataError((SQLException) t) ? invalidFeatureData(t) : error; + } } - RunnableStream mutationStream = - featureSqlSource - .via(featureWriter) - .to(Sink.ignore()) - .withResult((Builder) builder) - .handleError( - (result, throwable) -> { - Throwable error = throwable; - - if (throwable instanceof PSQLException - || throwable instanceof JsonParseException) { - error = - new IllegalArgumentException( - "Invalid feature data. You may be able to obtain more information about" - + " the problem by adding the header ‘Prefer: handling=strict’ to" - + " the request.", - throwable); - LogContext.errorAsDebug(LOGGER, throwable, "Error during feature mutation"); - } + return error; + } - return result.error(error); - }) - .handleItem((Builder::addIds)) - .handleEnd(Builder::build) - .on(getStreamRunner()); + private static Throwable invalidFeatureData(Throwable cause) { + LogContext.errorAsDebug(LOGGER, cause, "Error during feature mutation"); + + return new IllegalArgumentException( + "Invalid feature data. You may be able to obtain more information about the problem by" + + " adding the header \u2018Prefer: handling=strict\u2019 to the request.", + cause); + } + + /** + * The first SQLSTATE in the chain (causes and next-exceptions, e.g. of a batch failure) decides: + * connection failures (08), insufficient resources (53), operator intervention (57), system (58) + * and internal (XX) errors are not caused by the request; no SQLSTATE at all (e.g. a pool + * timeout) is not either. + */ + private static boolean isDataError(SQLException e) { + for (Throwable t : e) { + if (t instanceof SQLException && Objects.nonNull(((SQLException) t).getSQLState())) { + String state = ((SQLException) t).getSQLState(); + + return !(state.startsWith("08") + || state.startsWith("53") + || state.startsWith("57") + || state.startsWith("58") + || state.startsWith("XX")); + } + } - return mutationStream.run().toCompletableFuture().join(); + return false; } @Override diff --git a/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/domain/SqlClient.java b/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/domain/SqlClient.java index c0a79aad3..9aed97e84 100644 --- a/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/domain/SqlClient.java +++ b/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/domain/SqlClient.java @@ -7,18 +7,11 @@ */ package de.ii.xtraplatform.features.sql.domain; -import de.ii.xtraplatform.features.domain.Tuple; -import de.ii.xtraplatform.features.sql.app.FeatureDataSql; import de.ii.xtraplatform.streams.domain.Reactive; -import de.ii.xtraplatform.streams.domain.Reactive.Transformer; import java.sql.Connection; import java.util.Collection; import java.util.List; -import java.util.Optional; import java.util.concurrent.CompletableFuture; -import java.util.function.Consumer; -import java.util.function.Function; -import java.util.function.Supplier; public interface SqlClient extends SqlClientBasic { @@ -26,18 +19,6 @@ public interface SqlClient extends SqlClientBasic { Reactive.Source getSourceStream(String query, SqlQueryOptions options); - Reactive.Source getMutationSource( - List> statements, - List> idConsumers, - Object executionContext, - Optional featureId); - - Transformer getMutationFlow( - Function>>>> mutations, - Object executionContext, - String primaryKey, - Optional id); - List getNotifications(Connection connection); /** diff --git a/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/infra/db/JdbcSqlSession.java b/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/infra/db/JdbcSqlSession.java index 2b6ec06e1..138f40e8f 100644 --- a/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/infra/db/JdbcSqlSession.java +++ b/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/infra/db/JdbcSqlSession.java @@ -36,6 +36,7 @@ class JdbcSqlSession implements SqlSession { private final Connection connection; private boolean finalised; + private boolean released; private Savepoint activeSavepoint; // Non-fatal SQL warnings (e.g. PostgreSQL RAISE WARNING / RAISE NOTICE) emitted by mutation // statements, accumulated until the caller drains them. @@ -363,10 +364,12 @@ public void commit() { } try { connection.commit(); - finalised = true; } catch (SQLException e) { throw new IllegalStateException("Commit failed: " + e.getMessage(), e); } finally { + // a failed COMMIT ends the transaction as well (the database has rolled it back), so the + // session is finalised either way and must not attempt a rollback on close() + finalised = true; releaseConnection(); } } @@ -395,12 +398,24 @@ public void close() { } } + /** + * Returns the connection to the pool exactly once. Every step is isolated: on a connection that + * the database has terminated, resetting autocommit throws, and the lease must still be given + * back, otherwise the pool loses a connection for good. + */ private void releaseConnection() { + if (released) { + return; + } + released = true; + try { - if (!connection.isClosed()) { - connection.setAutoCommit(true); - connection.close(); - } + connection.setAutoCommit(true); + } catch (SQLException e) { + LOGGER.debug("Resetting autocommit failed: {}", e.getMessage()); + } + try { + connection.close(); } catch (SQLException e) { LOGGER.debug("Connection close failed: {}", e.getMessage()); } diff --git a/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/infra/db/MutationTransactionGuard.java b/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/infra/db/MutationTransactionGuard.java deleted file mode 100644 index e726dc8e8..000000000 --- a/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/infra/db/MutationTransactionGuard.java +++ /dev/null @@ -1,106 +0,0 @@ -/* - * Copyright 2026 interactive instruments GmbH - * - * This Source Code Form is subject to the terms of the Mozilla Public - * License, v. 2.0. If a copy of the MPL was not distributed with this - * file, You can obtain one at http://mozilla.org/MPL/2.0/. - */ -package de.ii.xtraplatform.features.sql.infra.db; - -import de.ii.xtraplatform.base.domain.LogContext; -import io.reactivex.rxjava3.core.Flowable; -import java.sql.Connection; -import java.sql.ResultSet; -import java.sql.SQLException; -import java.sql.Statement; -import java.util.Objects; -import java.util.concurrent.atomic.AtomicInteger; -import java.util.concurrent.atomic.AtomicReference; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -/** - * Reference accounting for a single rxjava3-jdbc transacted mutation chain. - * - *

The connection behind such a chain is reference counted: the count starts at 1, every {@code - * tx.update(...)} forks it (+1), and every statement stream that terminates commits or rolls back - * (-1). The real COMMIT/ROLLBACK and the real close (which returns the connection to the pool) - * happen only when the count reaches 0. - * - *

A statement stream that is cancelled instead of terminated fires neither handler, so its - * reference is never released. That is exactly what happens to all preceding statements when a - * later statement of the chain fails: the count never reaches 0, the transaction is neither - * committed nor rolled back, and the pooled connection stays leased forever. - * - *

This guard mirrors the count from the outside: {@link #acquired()} for every reference taken, - * {@link #track(Flowable)} to release it again on any terminal event, and {@link - * #releaseIfLeaked()} at the end of the chain to drain whatever is left. - */ -class MutationTransactionGuard { - - private static final Logger LOGGER = LoggerFactory.getLogger(MutationTransactionGuard.class); - - private final AtomicInteger outstanding = new AtomicInteger(0); - private final AtomicReference connection = new AtomicReference<>(); - - /** A reference was taken, either the initial connection or a fork from {@code tx.update(...)}. */ - void acquired() { - outstanding.incrementAndGet(); - } - - /** A reference is released on any terminal event, but not on cancellation. */ - Flowable track(Flowable stage) { - return stage - .doOnComplete(outstanding::decrementAndGet) - .doOnError(throwable -> outstanding.decrementAndGet()); - } - - /** The transacted connection is only reachable through a running statement's result set. */ - void capture(ResultSet resultSet) throws SQLException { - if (Objects.isNull(connection.get())) { - Statement statement = resultSet.getStatement(); - - if (Objects.nonNull(statement)) { - connection.compareAndSet(null, statement.getConnection()); - } - } - } - - /** Drains references left behind by cancelled statement streams, no-op on the happy path. */ - void releaseIfLeaked() { - int leaked = outstanding.getAndSet(0); - - if (leaked <= 0) { - return; - } - - Connection con = connection.get(); - - if (Objects.isNull(con)) { - LOGGER.debug( - "Abandoned mutation transaction with {} unreleased reference(s), no connection captured", - leaked); - return; - } - - LOGGER.warn( - "Abandoned mutation transaction, rolling back and releasing the connection ({} unreleased reference(s))", - leaked); - - try { - for (int i = 0; i < leaked && !con.isClosed(); i++) { - // performs the real ROLLBACK once the reference count reaches 0 - con.rollback(); - // no-op until the reference count is 0, then returns the connection to the pool - con.close(); - } - - if (!con.isClosed()) { - LOGGER.error( - "Could not release the connection of an abandoned mutation transaction, the connection pool may become depleted"); - } - } catch (SQLException e) { - LogContext.errorAsWarn(LOGGER, e, "Error releasing an abandoned mutation transaction"); - } - } -} diff --git a/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/infra/db/SqlClientRx.java b/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/infra/db/SqlClientRx.java index 2c8e67b6d..89a6a2a93 100644 --- a/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/infra/db/SqlClientRx.java +++ b/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/infra/db/SqlClientRx.java @@ -9,10 +9,7 @@ import com.google.common.collect.ImmutableList; import com.zaxxer.hikari.pool.ProxyConnection; -import de.ii.xtraplatform.base.domain.LogContext; import de.ii.xtraplatform.base.domain.LogContext.MARKER; -import de.ii.xtraplatform.features.domain.Tuple; -import de.ii.xtraplatform.features.sql.app.FeatureDataSql; import de.ii.xtraplatform.features.sql.domain.SqlClient; import de.ii.xtraplatform.features.sql.domain.SqlDbmsAdapter; import de.ii.xtraplatform.features.sql.domain.SqlDialect; @@ -20,11 +17,9 @@ import de.ii.xtraplatform.features.sql.domain.SqlRow; import de.ii.xtraplatform.features.sql.domain.SqlSession; import de.ii.xtraplatform.streams.domain.Reactive; -import de.ii.xtraplatform.streams.domain.Reactive.Transformer; import io.reactivex.rxjava3.core.Flowable; import io.reactivex.rxjava3.schedulers.Schedulers; import java.sql.Connection; -import java.sql.ResultSet; import java.sql.SQLException; import java.text.Collator; import java.util.ArrayList; @@ -35,13 +30,9 @@ import java.util.Optional; import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; -import java.util.function.Consumer; -import java.util.function.Function; -import java.util.function.Supplier; import java.util.stream.Collectors; import java.util.stream.Stream; import org.davidmoten.rxjava3.jdbc.Database; -import org.davidmoten.rxjava3.jdbc.Tx; import org.davidmoten.rxjava3.jdbc.internal.DelegatedConnection; import org.postgresql.PGConnection; import org.postgresql.PGNotification; @@ -201,190 +192,6 @@ public Reactive.Source getSourceStream(String query, SqlQueryOptions opt return Reactive.Source.publisher(flowable); } - @Override - public Reactive.Source getMutationSource( - List> statements, - List> idConsumers, - Object executionContext, - Optional featureId) { - /*List> toStatementsWithLog = - statements.stream() - .map( - function -> - (Function) - featureSql -> { - String statement = function.apply(featureSql); - - if (LOGGER.isDebugEnabled(MARKER.SQL)) { - LOGGER.debug(MARKER.SQL, "Executing statement: {}", statement); - } - - return statement; - }) - .collect(Collectors.toList());*/ - - // rxjava3-jdbc does not release the transacted connection when a statement stream is cancelled, - // which is what happens to all preceding statements when a later one fails, see the guard - MutationTransactionGuard guard = new MutationTransactionGuard(); - - String first = statements.get(0).get(); - if (LOGGER.isDebugEnabled(MARKER.SQL)) { - LOGGER.debug(MARKER.SQL, "Executing statement: {}", first); - } - - Flowable> txFlowable = - guard - .track( - session - .update(first) - .transacted() - .returnGeneratedKeys() - .get( - resultSet -> { - guard.capture(resultSet); - return consumeId(resultSet, null, idConsumers, 0); - }) - // the transacted connection is created when the statement stream is subscribed - .doOnSubscribe(subscription -> guard.acquired())) - .filter(tx -> !tx.isComplete()); - - for (int j = 1; j < statements.size(); j++) { - int finalJ = j; - txFlowable = - txFlowable.flatMap( - tx -> { - String next = statements.get(finalJ).get(); - if (LOGGER.isDebugEnabled(MARKER.SQL)) { - LOGGER.debug(MARKER.SQL, "Executing statement: {}", next); - } - - // tx.update forks the transacted connection - guard.acquired(); - - return guard - .track( - tx.update(next) - .returnGeneratedKeys() - .get( - resultSet -> { - guard.capture(resultSet); - return consumeId( - resultSet, - tx.value() instanceof String ? (String) tx.value() : null, - idConsumers, - finalJ); - })) - .filter(tx2 -> !tx2.isComplete()); - }); - } - - Flowable flowable = - txFlowable - .map(tx -> featureId.orElse((String) tx.value())) - .doFinally(guard::releaseIfLeaked); - - return Reactive.Source.publisher(flowable); - } - - private static String consumeId( - ResultSet resultSet, String previousId, List> idConsumers, int index) { - // null not allowed as return value - String id = null; - - try { - id = resultSet.getString(1); - - if (index < idConsumers.size()) { - Consumer idConsumer = idConsumers.get(index); - - if (Objects.nonNull(idConsumer)) { - idConsumer.accept(id); - } - } else if (LOGGER.isWarnEnabled()) { - LOGGER.warn("No id consumer for mutation statement {}, returned id: {}", index, id); - } - } catch (SQLException e) { - LogContext.errorAsDebug( - LOGGER, e, "Could not read the id returned by mutation statement {}", index); - } - - return previousId != null ? previousId : id; - } - - @Override - public Transformer getMutationFlow( - Function>>>> mutations, - Object executionContext, - String primaryKey, - Optional id) { - - Reactive.Transformer toQueries = - Reactive.Transformer.flatMap( - feature -> { - List>>> m = mutations.apply(feature); - - // both lists have to stay index aligned, the statements are resolved lazily since - // they may depend on ids returned by preceding statements - List> statements = new ArrayList<>(); - List> idConsumers = new ArrayList<>(); - - for (Supplier>> queryFunction : m) { - Tuple> query = queryFunction.get(); - - if (Objects.isNull(query.first())) { - continue; - } - - statements.add(() -> queryFunction.get().first()); - idConsumers.add(query.second()); - } - - Optional featureId = - feature - .getMapping() - .getColumnForId() - .flatMap( - idCol -> { - if (!Objects.equals(primaryKey, idCol.second().getName()) - && feature - .getRows() - .get(0) - .first() - .getFullPath() - .equals(idCol.first().getFullPath())) { - return Optional.ofNullable( - feature - .getRows() - .get(0) - .second() - .getValues() - .get(idCol.second().getName())); - } - return Optional.empty(); - }) - .map(SqlClientRx::unquote); - - return getMutationSource(statements, idConsumers, executionContext, featureId); - }); - - if (id.isPresent()) { - // TODO: check that feature id equals given id - Reactive.Transformer filter = - Reactive.Transformer.filter(featureSql -> true); - - return filter.via(toQueries); - } - - return toQueries; - } - - private static String unquote(String value) { - if (value.startsWith("'") && value.endsWith("'")) { - return value.substring(1, value.length() - 1); - } - return value; - } - @Override public Connection getConnection() { return session.connection().blockingGet(); diff --git a/xtraplatform-features-sql/src/test/groovy/de/ii/xtraplatform/features/sql/app/FeatureMutationsSqlSpec.groovy b/xtraplatform-features-sql/src/test/groovy/de/ii/xtraplatform/features/sql/app/FeatureMutationsSqlSpec.groovy index fd081f9c9..3af3ce237 100644 --- a/xtraplatform-features-sql/src/test/groovy/de/ii/xtraplatform/features/sql/app/FeatureMutationsSqlSpec.groovy +++ b/xtraplatform-features-sql/src/test/groovy/de/ii/xtraplatform/features/sql/app/FeatureMutationsSqlSpec.groovy @@ -33,7 +33,7 @@ class FeatureMutationsSqlSpec extends Specification { given: - FeatureMutationsSql inserts = Spy(new FeatureMutationsSql(null, new SqlInsertGenerator2(OgcCrs.CRS84, null, new ImmutableSqlPathDefaults.Builder().build()),new ImmutableSqlPathDefaults.Builder().build())) + FeatureMutationsSql inserts = Spy(new FeatureMutationsSql(new SqlInsertGenerator2(OgcCrs.CRS84, null, new ImmutableSqlPathDefaults.Builder().build()),new ImmutableSqlPathDefaults.Builder().build())) Map, List> rows = ImmutableMap., List> builder() .put(MAIN_M_2_N_SCHEMA.getFullPath(), ImmutableList.of(3)) @@ -85,7 +85,7 @@ class FeatureMutationsSqlSpec extends Specification { given: FeatureStoreInsertGenerator generator = Mock(); - FeatureMutationsSql inserts = new FeatureMutationsSql(null, generator,null) + FeatureMutationsSql inserts = new FeatureMutationsSql(generator, null) List rows = ImmutableList.of(0, 0, 1) when: @@ -104,7 +104,7 @@ class FeatureMutationsSqlSpec extends Specification { given: FeatureStoreInsertGenerator generator = Mock(); - FeatureMutationsSql inserts = new FeatureMutationsSql(null, generator, null) + FeatureMutationsSql inserts = new FeatureMutationsSql(generator, null) List rows = ImmutableList.of(0, 0, 0, 1) when: @@ -127,7 +127,7 @@ class FeatureMutationsSqlSpec extends Specification { given: FeatureStoreInsertGenerator generator = Mock(); - FeatureMutationsSql inserts = new FeatureMutationsSql(null, generator, null) + FeatureMutationsSql inserts = new FeatureMutationsSql(generator, null) List rows = ImmutableList.of(0, 0, 0, 1) when: diff --git a/xtraplatform-features-sql/src/test/groovy/de/ii/xtraplatform/features/sql/infra/db/JdbcSqlSessionSpec.groovy b/xtraplatform-features-sql/src/test/groovy/de/ii/xtraplatform/features/sql/infra/db/JdbcSqlSessionSpec.groovy index 94d163339..82cb302e9 100644 --- a/xtraplatform-features-sql/src/test/groovy/de/ii/xtraplatform/features/sql/infra/db/JdbcSqlSessionSpec.groovy +++ b/xtraplatform-features-sql/src/test/groovy/de/ii/xtraplatform/features/sql/infra/db/JdbcSqlSessionSpec.groovy @@ -457,4 +457,51 @@ class JdbcSqlSessionSpec extends Specification { session.rollback() 1 * connection.rollback() // only the original attempt } + + def 'a failed commit surfaces the cause, releases the connection and finalises the session'() { + given: + def session = new JdbcSqlSession(connection) + connection.commit() >> { throw new SQLException('violates deferred foreign key constraint', '23503') } + + when: + session.commit() + + then: + thrown(IllegalStateException) + 1 * connection.close() + + when: 'the caller cleans up as usual' + session.rollback() + session.close() + + then: 'the database has already ended the transaction, nothing is rolled back or released twice' + 0 * connection.rollback() + 0 * connection.close() + } + + def 'a failed rollback still releases the connection'() { + given: + def session = new JdbcSqlSession(connection) + connection.rollback() >> { throw new SQLException('rolling-back-failed') } + + when: + session.rollback() + + then: + 1 * connection.close() + } + + def 'a connection terminated by the database is still returned to the pool'() { + given: + def session = new JdbcSqlSession(connection) + connection.rollback() >> { throw new SQLException('This connection has been closed.', '08003') } + connection.setAutoCommit(true) >> { throw new SQLException('This connection has been closed.', '08003') } + + when: + session.close() + + then: + noExceptionThrown() + 1 * connection.close() + } } diff --git a/xtraplatform-features-sql/src/test/groovy/de/ii/xtraplatform/features/sql/infra/db/SqlClientRxSpec.groovy b/xtraplatform-features-sql/src/test/groovy/de/ii/xtraplatform/features/sql/infra/db/SqlClientRxSpec.groovy deleted file mode 100644 index b0fd56b88..000000000 --- a/xtraplatform-features-sql/src/test/groovy/de/ii/xtraplatform/features/sql/infra/db/SqlClientRxSpec.groovy +++ /dev/null @@ -1,206 +0,0 @@ -/* - * Copyright 2026 interactive instruments GmbH - * - * This Source Code Form is subject to the terms of the Mozilla Public - * License, v. 2.0. If a copy of the MPL was not distributed with this - * file, You can obtain one at http://mozilla.org/MPL/2.0/. - */ -package de.ii.xtraplatform.features.sql.infra.db - -import de.ii.xtraplatform.features.sql.domain.SqlDbmsAdapter -import de.ii.xtraplatform.features.sql.domain.SqlDialect -import de.ii.xtraplatform.streams.app.SourceDefault -import io.reactivex.rxjava3.core.Flowable -import io.reactivex.rxjava3.subscribers.TestSubscriber -import org.davidmoten.rxjava3.jdbc.ConnectionProvider -import org.davidmoten.rxjava3.jdbc.Database -import spock.lang.Specification - -import java.sql.Connection -import java.sql.PreparedStatement -import java.sql.ResultSet -import java.sql.SQLException -import java.sql.Statement -import java.util.function.Consumer -import java.util.function.Supplier - -/** - * Locks the connection lifecycle of the rxjava3-jdbc mutation chain built by - * {@link SqlClientRx#getMutationSource}. - * - *

Why this matters: the transacted connection is reference counted and is only really rolled - * back and returned to the pool when the count reaches 0. A statement stream that is cancelled - * instead of terminated never releases its reference, and that is exactly what happens to all - * preceding statements when a later statement of the chain fails. Without the guard, every failed - * CREATE/PUT/PATCH whose error occurs after the first statement permanently leaks a pooled - * connection until the pool is starved (#1711). - */ -class SqlClientRxSpec extends Specification { - - Connection connection - Map preparedStatements - boolean closed - - SqlClientRx sqlClient - - def setup() { - preparedStatements = [:] - closed = false - - connection = Mock(Connection) - connection.getAutoCommit() >> false - connection.isClosed() >> { closed } - connection.prepareStatement(_ as String, Statement.RETURN_GENERATED_KEYS) >> { String sql, int keys -> - preparedStatements.get(sql) - } - - Database database = Database.fromBlocking(new ConnectionProvider() { - @Override - Connection get() { - return connection - } - - @Override - void close() { - } - }) - - sqlClient = new SqlClientRx(database, Mock(SqlDbmsAdapter), Mock(SqlDialect), Optional.empty()) - } - - def 'a failure in the second statement rolls back and releases the connection'() { - given: - statement('INSERT 1', 'id1') - failingStatement('INSERT 2') - - when: - TestSubscriber subscriber = subscribe(['INSERT 1', 'INSERT 2']) - - then: - subscriber.assertError(SQLException) - 1 * connection.rollback() - 1 * connection.close() >> { closed = true } - 0 * connection.commit() - } - - def 'a failure in the third statement releases both outstanding references'() { - given: - statement('INSERT 1', 'id1') - statement('INSERT 2', 'id2') - failingStatement('INSERT 3') - - when: - TestSubscriber subscriber = subscribe(['INSERT 1', 'INSERT 2', 'INSERT 3']) - - then: - subscriber.assertError(SQLException) - 1 * connection.rollback() - 1 * connection.close() >> { closed = true } - 0 * connection.commit() - } - - def 'a successful chain still commits exactly once and is not rolled back'() { - given: - statement('INSERT 1', 'id1') - statement('INSERT 2', 'id2') - - when: - TestSubscriber subscriber = subscribe(['INSERT 1', 'INSERT 2']) - - then: - subscriber.assertComplete() - 1 * connection.commit() - 1 * connection.close() >> { closed = true } - 0 * connection.rollback() - } - - def 'cancellation by the consumer rolls back and releases the connection'() { - given: - statement('INSERT 1', 'id1') - statement('INSERT 2', 'id2') - - when: - cancelAfterFirst(['INSERT 1', 'INSERT 2']) - - then: - 1 * connection.rollback() - 1 * connection.close() >> { closed = true } - 0 * connection.commit() - } - - def 'each id consumer receives the id returned by its own statement'() { - given: - statement('INSERT 1', 'id1') - statement('INSERT 2', 'id2') - statement('INSERT 3', 'id3') - - and: - List received = [null, null, null] - List> idConsumers = (0..2).collect { int index -> - ({ String id -> received.set(index, id) } as Consumer) - } - - when: - TestSubscriber subscriber = subscribe(['INSERT 1', 'INSERT 2', 'INSERT 3'], idConsumers) - - then: - subscriber.assertComplete() - received == ['id1', 'id2', 'id3'] - 1 * connection.close() >> { closed = true } - } - - def 'a missing id consumer does not fail the mutation'() { - given: - statement('INSERT 1', 'id1') - statement('INSERT 2', 'id2') - - when: - TestSubscriber subscriber = subscribe(['INSERT 1', 'INSERT 2'], [null, null]) - - then: - subscriber.assertComplete() - 1 * connection.commit() - 1 * connection.close() >> { closed = true } - 0 * connection.rollback() - } - - private void statement(String sql, String id) { - PreparedStatement preparedStatement = Mock(PreparedStatement) - ResultSet resultSet = Mock(ResultSet) - - preparedStatement.execute() >> true - preparedStatement.getGeneratedKeys() >> resultSet - resultSet.next() >>> [true, false] - resultSet.getString(1) >> id - - preparedStatements.put(sql, preparedStatement) - } - - private void failingStatement(String sql) { - PreparedStatement preparedStatement = Mock(PreparedStatement) - - preparedStatement.execute() >> { - throw new SQLException('duplicate key value violates unique constraint') - } - - preparedStatements.put(sql, preparedStatement) - } - - private TestSubscriber subscribe(List sql, List> idConsumers = null) { - return Flowable.fromPublisher(publisher(sql, idConsumers)).test() - } - - private void cancelAfterFirst(List sql) { - Flowable.fromPublisher(publisher(sql, null)).take(1).test() - } - - private org.reactivestreams.Publisher publisher(List sql, List> idConsumers) { - List> statements = sql.collect { String s -> ({ -> s } as Supplier) } - List> consumers = idConsumers ?: sql.collect { ({ String id -> } as Consumer) } - - SourceDefault source = - (SourceDefault) sqlClient.getMutationSource(statements, consumers, null, Optional.empty()) - - return source.getPublisher() - } -} From 8ae048a6c6002be8967bb4e4c0ee041548c76cd6 Mon Sep 17 00:00:00 2001 From: Clemens Portele Date: Sat, 29 Aug 2026 14:13:48 +0200 Subject: [PATCH 2/2] features/sql: lease connections from the pool without rxjava3-jdbc Sessions and statements without a result (creating result-set tables, routes DDL) obtained their connection through the rxjava3-jdbc Database. They now lease it from the pool directly and run as plain JDBC, so rxjava3-jdbc is confined to the streamed reads; a connection that cannot be leased is reported with the pool's message as the cause. --- .../features/sql/infra/db/SqlClientRx.java | 39 +++++-- .../features/sql/infra/db/SqlConnectorRx.java | 1 + .../sql/infra/db/SqlClientRxSpec.groovy | 106 ++++++++++++++++++ 3 files changed, 135 insertions(+), 11 deletions(-) create mode 100644 xtraplatform-features-sql/src/test/groovy/de/ii/xtraplatform/features/sql/infra/db/SqlClientRxSpec.groovy diff --git a/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/infra/db/SqlClientRx.java b/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/infra/db/SqlClientRx.java index 89a6a2a93..7eb721552 100644 --- a/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/infra/db/SqlClientRx.java +++ b/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/infra/db/SqlClientRx.java @@ -21,6 +21,7 @@ import io.reactivex.rxjava3.schedulers.Schedulers; import java.sql.Connection; import java.sql.SQLException; +import java.sql.Statement; import java.text.Collator; import java.util.ArrayList; import java.util.Arrays; @@ -32,8 +33,8 @@ import java.util.concurrent.TimeUnit; import java.util.stream.Collectors; import java.util.stream.Stream; +import javax.sql.DataSource; import org.davidmoten.rxjava3.jdbc.Database; -import org.davidmoten.rxjava3.jdbc.internal.DelegatedConnection; import org.postgresql.PGConnection; import org.postgresql.PGNotification; import org.slf4j.Logger; @@ -49,17 +50,22 @@ public class SqlClientRx implements SqlClient { // it exists to end. private static final long READ_STALL_TIMEOUT_MINUTES = 10; + // rxjava3-jdbc is used for streamed reads only; everything that needs a connection of its own + // (sessions, statements without a result) leases it from the pool directly private final Database session; + private final DataSource dataSource; private final SqlDbmsAdapter dbmsAdapter; private final SqlDialect dialect; private final Collator collator; public SqlClientRx( Database session, + DataSource dataSource, SqlDbmsAdapter dbmsAdapter, SqlDialect dialect, Optional defaultCollation) { this.session = session; + this.dataSource = dataSource; this.dbmsAdapter = dbmsAdapter; this.dialect = dialect; this.collator = dbmsAdapter.getRowSortingCollator(defaultCollation); @@ -73,10 +79,15 @@ public CompletableFuture> run(String query, SqlQueryOptions o CompletableFuture> result = new CompletableFuture<>(); if (options.getColumnTypes().isEmpty()) { - session - .update(query) - .complete() - .subscribe(() -> result.complete(ImmutableList.of()), result::completeExceptionally); + // a statement without a result (DDL, INSERT, DROP); autocommit is on, so it is committed when + // it returns and the connection goes back to the pool in every case + try (Connection connection = dataSource.getConnection(); + Statement statement = connection.createStatement()) { + statement.execute(query); + result.complete(ImmutableList.of()); + } catch (SQLException e) { + result.completeExceptionally(e); + } return result; } @@ -194,13 +205,22 @@ public Reactive.Source getSourceStream(String query, SqlQueryOptions opt @Override public Connection getConnection() { - return session.connection().blockingGet(); + return leaseConnection(); } @Override public SqlSession openSession() { - Connection connection = session.connection().blockingGet(); - return new JdbcSqlSession(connection); + return new JdbcSqlSession(leaseConnection()); + } + + /** A pooled connection; closing it returns it to the pool. */ + private Connection leaseConnection() { + try { + return dataSource.getConnection(); + } catch (SQLException e) { + throw new IllegalStateException( + "Could not obtain a database connection: " + e.getMessage(), e); + } } @Override @@ -217,9 +237,6 @@ public SqlDbmsAdapter getDbmsAdapter() { public List getNotifications(Connection connection) { Connection actualConnection = connection; - if (actualConnection instanceof DelegatedConnection) { - actualConnection = ((DelegatedConnection) actualConnection).con(); - } if (actualConnection instanceof ProxyConnection) { try { actualConnection = actualConnection.unwrap(Connection.class); diff --git a/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/infra/db/SqlConnectorRx.java b/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/infra/db/SqlConnectorRx.java index 37d054154..5a8a1432c 100644 --- a/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/infra/db/SqlConnectorRx.java +++ b/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/infra/db/SqlConnectorRx.java @@ -170,6 +170,7 @@ public void start() { this.sqlClient = new SqlClientRx( session, + dataSource, dbmsAdapters.get(connectionInfo.getDialect()), dbmsAdapters.getDialect(connectionInfo.getDialect()), connectionInfo.getDefaultCollation()); diff --git a/xtraplatform-features-sql/src/test/groovy/de/ii/xtraplatform/features/sql/infra/db/SqlClientRxSpec.groovy b/xtraplatform-features-sql/src/test/groovy/de/ii/xtraplatform/features/sql/infra/db/SqlClientRxSpec.groovy new file mode 100644 index 000000000..f8db01dd4 --- /dev/null +++ b/xtraplatform-features-sql/src/test/groovy/de/ii/xtraplatform/features/sql/infra/db/SqlClientRxSpec.groovy @@ -0,0 +1,106 @@ +/* + * Copyright 2026 interactive instruments GmbH + * + * This Source Code Form is subject to the terms of the Mozilla Public + * License, v. 2.0. If a copy of the MPL was not distributed with this + * file, You can obtain one at http://mozilla.org/MPL/2.0/. + */ +package de.ii.xtraplatform.features.sql.infra.db + +import de.ii.xtraplatform.features.sql.domain.SqlDbmsAdapter +import de.ii.xtraplatform.features.sql.domain.SqlDialect +import de.ii.xtraplatform.features.sql.domain.SqlQueryOptions +import org.davidmoten.rxjava3.jdbc.ConnectionProvider +import org.davidmoten.rxjava3.jdbc.Database +import spock.lang.Specification + +import javax.sql.DataSource +import java.sql.Connection +import java.sql.SQLException +import java.sql.Statement +import java.util.concurrent.CompletionException + +/** + * Locks the connection lifecycle of everything in {@link SqlClientRx} that is not a streamed read: + * a connection is leased from the pool directly and returned to it in every case, so no code path + * outside the reads depends on the rxjava3-jdbc reference counting. + */ +class SqlClientRxSpec extends Specification { + + DataSource dataSource + Connection connection + Statement statement + boolean leaseFails + + SqlClientRx sqlClient + + def setup() { + dataSource = Mock(DataSource) + connection = Mock(Connection) + statement = Mock(Statement) + leaseFails = false + dataSource.getConnection() >> { + if (leaseFails) throw new SQLException('Connection is not available, request timed out after 30000ms') + return connection + } + connection.createStatement() >> statement + + Database database = Database.fromBlocking(new ConnectionProvider() { + @Override + Connection get() { return connection } + + @Override + void close() {} + }) + + sqlClient = new SqlClientRx(database, dataSource, Mock(SqlDbmsAdapter), Mock(SqlDialect), Optional.empty()) + } + + def 'a statement without a result runs on a pooled connection that is closed afterwards'() { + when: + def result = sqlClient.run('CREATE TABLE t (id int)', SqlQueryOptions.ddl()).join() + + then: + result.isEmpty() + 1 * statement.execute('CREATE TABLE t (id int)') >> false + 1 * statement.close() + 1 * connection.close() + } + + def 'a failing statement without a result surfaces the error and still closes the connection'() { + given: + statement.execute(_ as String) >> { throw new SQLException('relation does not exist', '42P01') } + + when: + sqlClient.run('DROP TABLE t', SqlQueryOptions.ddl()).join() + + then: + def e = thrown(CompletionException) + e.cause instanceof SQLException + 1 * connection.close() + } + + def 'a session leases its connection from the pool and returns it on close'() { + when: + def session = sqlClient.openSession() + session.close() + + then: + 1 * dataSource.getConnection() >> connection + 1 * connection.setAutoCommit(false) + 1 * connection.close() + } + + def 'a connection that cannot be leased is reported with its cause'() { + given: + leaseFails = true + + when: + sqlClient.getConnection() + + then: + def e = thrown(IllegalStateException) + e.cause instanceof SQLException + e.message.contains('not available') + } +}