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..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 @@ -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,12 +17,11 @@ 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.sql.Statement; import java.text.Collator; import java.util.ArrayList; import java.util.Arrays; @@ -35,14 +31,10 @@ 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 javax.sql.DataSource; 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; import org.slf4j.Logger; @@ -58,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); @@ -82,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; } @@ -202,198 +204,23 @@ public Reactive.Source getSourceStream(String query, SqlQueryOptions opt } @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; + public Connection getConnection() { + return leaseConnection(); } @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; + public SqlSession openSession() { + return new JdbcSqlSession(leaseConnection()); } - private static String unquote(String value) { - if (value.startsWith("'") && value.endsWith("'")) { - return value.substring(1, value.length() - 1); + /** 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); } - return value; - } - - @Override - public Connection getConnection() { - return session.connection().blockingGet(); - } - - @Override - public SqlSession openSession() { - Connection connection = session.connection().blockingGet(); - return new JdbcSqlSession(connection); } @Override @@ -410,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/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 index b0fd56b88..f8db01dd4 100644 --- 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 @@ -9,198 +9,98 @@ 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 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.PreparedStatement -import java.sql.ResultSet import java.sql.SQLException import java.sql.Statement -import java.util.function.Consumer -import java.util.function.Supplier +import java.util.concurrent.CompletionException /** - * 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). + * 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 - Map preparedStatements - boolean closed + Statement statement + boolean leaseFails SqlClientRx sqlClient def setup() { - preparedStatements = [:] - closed = false - + dataSource = Mock(DataSource) connection = Mock(Connection) - connection.getAutoCommit() >> false - connection.isClosed() >> { closed } - connection.prepareStatement(_ as String, Statement.RETURN_GENERATED_KEYS) >> { String sql, int keys -> - preparedStatements.get(sql) + 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 - } + Connection get() { return connection } @Override - void close() { - } + void close() {} }) - sqlClient = new SqlClientRx(database, Mock(SqlDbmsAdapter), Mock(SqlDialect), Optional.empty()) + sqlClient = new SqlClientRx(database, dataSource, 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') - + def 'a statement without a result runs on a pooled connection that is closed afterwards'() { when: - TestSubscriber subscriber = subscribe(['INSERT 1', 'INSERT 2']) + def result = sqlClient.run('CREATE TABLE t (id int)', SqlQueryOptions.ddl()).join() then: - subscriber.assertError(SQLException) - 1 * connection.rollback() - 1 * connection.close() >> { closed = true } - 0 * connection.commit() + result.isEmpty() + 1 * statement.execute('CREATE TABLE t (id int)') >> false + 1 * statement.close() + 1 * connection.close() } - def 'a failure in the third statement releases both outstanding references'() { + def 'a failing statement without a result surfaces the error and still closes the connection'() { given: - statement('INSERT 1', 'id1') - statement('INSERT 2', 'id2') - failingStatement('INSERT 3') + statement.execute(_ as String) >> { throw new SQLException('relation does not exist', '42P01') } when: - TestSubscriber subscriber = subscribe(['INSERT 1', 'INSERT 2', 'INSERT 3']) + sqlClient.run('DROP TABLE t', SqlQueryOptions.ddl()).join() then: - subscriber.assertError(SQLException) - 1 * connection.rollback() - 1 * connection.close() >> { closed = true } - 0 * connection.commit() + def e = thrown(CompletionException) + e.cause instanceof SQLException + 1 * connection.close() } - 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) - } - + def 'a session leases its connection from the pool and returns it on close'() { when: - TestSubscriber subscriber = subscribe(['INSERT 1', 'INSERT 2', 'INSERT 3'], idConsumers) + def session = sqlClient.openSession() + session.close() then: - subscriber.assertComplete() - received == ['id1', 'id2', 'id3'] - 1 * connection.close() >> { closed = true } + 1 * dataSource.getConnection() >> connection + 1 * connection.setAutoCommit(false) + 1 * connection.close() } - def 'a missing id consumer does not fail the mutation'() { + def 'a connection that cannot be leased is reported with its cause'() { given: - statement('INSERT 1', 'id1') - statement('INSERT 2', 'id2') + leaseFails = true when: - TestSubscriber subscriber = subscribe(['INSERT 1', 'INSERT 2'], [null, null]) + sqlClient.getConnection() 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() + def e = thrown(IllegalStateException) + e.cause instanceof SQLException + e.message.contains('not available') } }