diff --git a/vertx-db2-client/src/main/java/io/vertx/db2client/impl/DB2SocketConnection.java b/vertx-db2-client/src/main/java/io/vertx/db2client/impl/DB2SocketConnection.java index 69171735e..95292847d 100644 --- a/vertx-db2-client/src/main/java/io/vertx/db2client/impl/DB2SocketConnection.java +++ b/vertx-db2-client/src/main/java/io/vertx/db2client/impl/DB2SocketConnection.java @@ -32,6 +32,7 @@ import io.vertx.sqlclient.spi.connection.Connection; import io.vertx.sqlclient.spi.protocol.CommandBase; import io.vertx.sqlclient.spi.protocol.ExtendedQueryCommand; +import io.vertx.sqlclient.spi.protocol.SavepointCommand; import io.vertx.sqlclient.spi.protocol.SimpleQueryCommand; import io.vertx.sqlclient.spi.protocol.TxCommand; @@ -121,11 +122,27 @@ protected void doSchedule(CommandBase cmd, Completable handler) { super.doSchedule(cmd2, (res, err) -> handler.complete(txCmd.result(), err)); } + } else if (cmd instanceof SavepointCommand) { + SavepointCommand savepoint = (SavepointCommand) cmd; + SimpleQueryCommand cmd2 = new SimpleQueryCommand<>(savepointSql(savepoint), false, false, + SocketConnectionBase.NULL_COLLECTOR, QueryResultHandler.NOOP_HANDLER); + super.doSchedule(cmd2, (res, err) -> handler.complete(savepoint.result(), err)); } else { super.doSchedule(cmd, handler); } } + /** + * DB2 requires the {@code ON ROLLBACK RETAIN CURSORS} clause when a savepoint is + * created, the other statements follow the standard syntax. + */ + private static String savepointSql(SavepointCommand savepoint) { + if (savepoint.kind() == SavepointCommand.Kind.CREATE) { + return "SAVEPOINT " + savepoint.name() + " ON ROLLBACK RETAIN CURSORS"; + } + return savepoint.sql(); + } + @Override public String system() { return "db2"; diff --git a/vertx-db2-client/src/main/java/io/vertx/db2client/spi/DB2Driver.java b/vertx-db2-client/src/main/java/io/vertx/db2client/spi/DB2Driver.java index e679cece9..350417246 100644 --- a/vertx-db2-client/src/main/java/io/vertx/db2client/spi/DB2Driver.java +++ b/vertx-db2-client/src/main/java/io/vertx/db2client/spi/DB2Driver.java @@ -82,6 +82,11 @@ public ConnectionFactory createConnectionFactory(Vertx vertx, return new DB2ConnectionFactory((VertxInternal) vertx, transportOptions); } + @Override + public boolean supportsSavepoints() { + return true; + } + @Override public SqlConnectionInternal wrapConnection(ContextInternal context, ConnectionFactory factory, Connection connection) { return new DB2ConnectionImpl(context, factory, connection); diff --git a/vertx-db2-client/src/test/java/io/vertx/tests/db2client/tck/DB2TransactionTest.java b/vertx-db2-client/src/test/java/io/vertx/tests/db2client/tck/DB2TransactionTest.java index 5383511db..e6369bf78 100644 --- a/vertx-db2-client/src/test/java/io/vertx/tests/db2client/tck/DB2TransactionTest.java +++ b/vertx-db2-client/src/test/java/io/vertx/tests/db2client/tck/DB2TransactionTest.java @@ -12,6 +12,9 @@ import io.vertx.db2client.DB2Builder; import io.vertx.db2client.DB2ConnectOptions; +import io.vertx.ext.unit.Async; +import io.vertx.sqlclient.Cursor; +import org.junit.Test; import io.vertx.ext.unit.TestContext; import io.vertx.ext.unit.junit.VertxUnitRunner; import io.vertx.sqlclient.Pool; @@ -73,4 +76,43 @@ protected void cleanTestTable(TestContext ctx) { protected String statement(String... parts) { return String.join("?", parts); } + + @Override + protected boolean supportsSavepoints() { + return true; + } + + /** + * DB2 creates savepoints with ON ROLLBACK RETAIN CURSORS, so a cursor opened before the + * rollback keeps returning the rows it was already positioned on. + */ + @Test + public void testCursorSurvivesRollbackToSavepoint(TestContext ctx) { + Async async = ctx.async(); + connector.accept(ctx.asyncAssertSuccess(res -> { + insertMutable(res.client, 1, "one") + .compose(v -> insertMutable(res.client, 2, "two")) + .compose(v -> res.client.prepare("SELECT id FROM mutable ORDER BY id")) + .compose(ps -> { + Cursor cursor = ps.cursor(); + return cursor.read(1) + .compose(first -> { + ctx.assertEquals(1, first.size()); + ctx.assertEquals(1, first.iterator().next().getInteger("id")); + return res.tx.createSavepoint(); + }) + .compose(sp -> insertMutable(res.client, 3, "three").compose(v -> sp.rollback())) + // the cursor was opened before the savepoint, it must still be readable + .compose(v -> cursor.read(1)) + .compose(second -> { + ctx.assertEquals(1, second.size()); + ctx.assertEquals(2, second.iterator().next().getInteger("id")); + return cursor.close(); + }); + }) + .compose(v -> res.tx.commit()) + .compose(v -> assertMutableIds(ctx, 1, 2)) + .onComplete(ctx.asyncAssertSuccess(v -> async.complete())); + })); + } } diff --git a/vertx-mssql-client/src/main/java/io/vertx/mssqlclient/impl/MSSQLSocketConnection.java b/vertx-mssql-client/src/main/java/io/vertx/mssqlclient/impl/MSSQLSocketConnection.java index 945deee1a..f85bef478 100644 --- a/vertx-mssql-client/src/main/java/io/vertx/mssqlclient/impl/MSSQLSocketConnection.java +++ b/vertx-mssql-client/src/main/java/io/vertx/mssqlclient/impl/MSSQLSocketConnection.java @@ -42,6 +42,8 @@ import java.util.Map; import java.util.function.Predicate; +import io.vertx.sqlclient.spi.protocol.SavepointCommand; + import static io.vertx.sqlclient.spi.protocol.TxCommand.Kind.BEGIN; public class MSSQLSocketConnection extends SocketConnectionBase { @@ -158,6 +160,20 @@ public void init() { return MSSQLCommandMessage.wrap(command); } + /** + * Transact-SQL names the savepoint statements differently and has no statement to + * release one, {@link io.vertx.sqlclient.spi.Driver#supportsSavepointRelease()} + * reports that. + */ + private static String savepointSql(SavepointCommand savepoint) { + switch (savepoint.kind()) { + case CREATE: + return "SAVE TRANSACTION " + savepoint.name(); + default: + return "ROLLBACK TRANSACTION " + savepoint.name(); + } + } + @Override protected void doSchedule(CommandBase cmd, Completable handler) { if (cmd instanceof TxCommand) { @@ -170,6 +186,21 @@ protected void doSchedule(CommandBase cmd, Completable handler) { SocketConnectionBase.NULL_COLLECTOR, QueryResultHandler.NOOP_HANDLER); super.doSchedule(cmd2, (res, err) -> handler.complete(tx.result(), err)); + } else if (cmd instanceof SavepointCommand) { + SavepointCommand savepoint = (SavepointCommand) cmd; + if (savepoint.kind() == SavepointCommand.Kind.RELEASE) { + // Guarded by MSSQLDriver#supportsSavepointRelease, fail rather than throw on the event loop + handler.fail(new UnsupportedOperationException( + "Releasing a savepoint is not supported by Microsoft SQL Server")); + return; + } + SimpleQueryCommand cmd2 = new SimpleQueryCommand<>( + savepointSql(savepoint), + false, + false, + SocketConnectionBase.NULL_COLLECTOR, + QueryResultHandler.NOOP_HANDLER); + super.doSchedule(cmd2, (res, err) -> handler.complete(savepoint.result(), err)); } else { super.doSchedule(cmd, handler); } diff --git a/vertx-mssql-client/src/main/java/io/vertx/mssqlclient/spi/MSSQLDriver.java b/vertx-mssql-client/src/main/java/io/vertx/mssqlclient/spi/MSSQLDriver.java index 7901059c5..7858b5c96 100644 --- a/vertx-mssql-client/src/main/java/io/vertx/mssqlclient/spi/MSSQLDriver.java +++ b/vertx-mssql-client/src/main/java/io/vertx/mssqlclient/spi/MSSQLDriver.java @@ -67,6 +67,19 @@ public int appendQueryPlaceholder(StringBuilder queryBuilder, int index, int cur return index; } + @Override + public boolean supportsSavepoints() { + return true; + } + + /** + * Transact-SQL has no statement that discards a savepoint without rolling back to it. + */ + @Override + public boolean supportsSavepointRelease() { + return false; + } + @Override public SqlConnectionInternal wrapConnection(ContextInternal context, ConnectionFactory factory, Connection connection) { return new MSSQLConnectionImpl(context, factory, connection); diff --git a/vertx-mssql-client/src/test/java/io/vertx/tests/mssqlclient/tck/MSSQLTransactionTest.java b/vertx-mssql-client/src/test/java/io/vertx/tests/mssqlclient/tck/MSSQLTransactionTest.java index f08608821..0e0cd09c0 100644 --- a/vertx-mssql-client/src/test/java/io/vertx/tests/mssqlclient/tck/MSSQLTransactionTest.java +++ b/vertx-mssql-client/src/test/java/io/vertx/tests/mssqlclient/tck/MSSQLTransactionTest.java @@ -10,7 +10,10 @@ */ package io.vertx.tests.mssqlclient.tck; +import io.vertx.core.Future; +import io.vertx.sqlclient.SqlConnection; import io.vertx.ext.unit.TestContext; +import io.vertx.ext.unit.Async; import io.vertx.ext.unit.junit.VertxUnitRunner; import io.vertx.mssqlclient.MSSQLBuilder; import io.vertx.mssqlclient.MSSQLConnectOptions; @@ -55,4 +58,56 @@ protected String statement(String... parts) { public void testDelayedCommit(TestContext ctx) { throw new AssumptionViolatedException("MSSQL holds write locks on inserted row with isolation level = 2"); } + + @Override + protected boolean supportsSavepoints() { + return true; + } + + @Override + protected boolean supportsSavepointRelease() { + return false; + } + + /** + * SQL Server drops the savepoint once the transaction has been rolled back to it. + */ + @Override + protected boolean supportsRepeatedRollbackToSavepoint() { + return false; + } + + /** + * A savepoint is a mark inside the current transaction, it must not open a nested one: + * @@TRANCOUNT stays at 1 after SAVE TRANSACTION and after rolling back to it. + */ + @Test + public void testSavepointDoesNotNestTheTransaction(TestContext ctx) { + Async async = ctx.async(); + connector.accept(ctx.asyncAssertSuccess(res -> { + trancount(res.client) + .compose(before -> { + ctx.assertEquals(1, before, "the transaction should be the only one open"); + return res.tx.createSavepoint(); + }) + .compose(sp -> trancount(res.client) + .compose(afterSave -> { + ctx.assertEquals(1, afterSave, "SAVE TRANSACTION must not nest a transaction"); + return insertMutable(res.client, 1, "rolled-back"); + }) + .compose(v -> sp.rollback()) + .compose(v -> trancount(res.client)) + .compose(afterRollback -> { + ctx.assertEquals(1, afterRollback, "rolling back to a savepoint must keep the transaction open"); + return insertMutable(res.client, 2, "kept"); + }) + .compose(v -> res.tx.commit())) + .compose(v -> assertMutableIds(ctx, 2)) + .onComplete(ctx.asyncAssertSuccess(v -> async.complete())); + })); + } + + private Future trancount(SqlConnection client) { + return client.query("SELECT @@TRANCOUNT AS c").execute().map(rows -> rows.iterator().next().getInteger("c")); + } } diff --git a/vertx-mysql-client/src/main/java/io/vertx/mysqlclient/impl/MySQLSocketConnection.java b/vertx-mysql-client/src/main/java/io/vertx/mysqlclient/impl/MySQLSocketConnection.java index 60c102add..c01c6c649 100644 --- a/vertx-mysql-client/src/main/java/io/vertx/mysqlclient/impl/MySQLSocketConnection.java +++ b/vertx-mysql-client/src/main/java/io/vertx/mysqlclient/impl/MySQLSocketConnection.java @@ -43,6 +43,7 @@ import io.vertx.sqlclient.codec.SocketConnectionBase; import io.vertx.sqlclient.spi.protocol.CommandBase; import io.vertx.sqlclient.spi.protocol.ExtendedQueryCommand; +import io.vertx.sqlclient.spi.protocol.SavepointCommand; import io.vertx.sqlclient.spi.protocol.SimpleQueryCommand; import io.vertx.sqlclient.spi.protocol.TxCommand; import io.vertx.sqlclient.spi.DatabaseMetadata; @@ -127,6 +128,15 @@ protected void doSchedule(CommandBase cmd, Completable handler) { SocketConnectionBase.NULL_COLLECTOR, QueryResultHandler.NOOP_HANDLER); super.doSchedule(cmd2, (res, err) -> handler.complete(tx.result(), err)); + } else if (cmd instanceof SavepointCommand) { + SavepointCommand savepoint = (SavepointCommand) cmd; + SimpleQueryCommand cmd2 = new SimpleQueryCommand<>( + savepoint.sql(), + false, + false, + SocketConnectionBase.NULL_COLLECTOR, + QueryResultHandler.NOOP_HANDLER); + super.doSchedule(cmd2, (res, err) -> handler.complete(savepoint.result(), err)); } else { super.doSchedule(cmd, handler); } diff --git a/vertx-mysql-client/src/main/java/io/vertx/mysqlclient/spi/MySQLDriver.java b/vertx-mysql-client/src/main/java/io/vertx/mysqlclient/spi/MySQLDriver.java index f326ae353..c9fe487f2 100644 --- a/vertx-mysql-client/src/main/java/io/vertx/mysqlclient/spi/MySQLDriver.java +++ b/vertx-mysql-client/src/main/java/io/vertx/mysqlclient/spi/MySQLDriver.java @@ -85,6 +85,11 @@ public ConnectionFactory createConnectionFactory(Vertx vert return new MySQLConnectionFactory((VertxInternal) vertx, transportOptions); } + @Override + public boolean supportsSavepoints() { + return true; + } + @Override public SqlConnectionInternal wrapConnection(ContextInternal context, ConnectionFactory factory, Connection connection) { return new MySQLConnectionImpl(context, factory, connection); diff --git a/vertx-mysql-client/src/test/java/io/vertx/tests/mysqlclient/tck/MySQLTransactionTest.java b/vertx-mysql-client/src/test/java/io/vertx/tests/mysqlclient/tck/MySQLTransactionTest.java index ccef02cfd..c8003a84c 100644 --- a/vertx-mysql-client/src/test/java/io/vertx/tests/mysqlclient/tck/MySQLTransactionTest.java +++ b/vertx-mysql-client/src/test/java/io/vertx/tests/mysqlclient/tck/MySQLTransactionTest.java @@ -44,4 +44,11 @@ protected Pool nonTxPool() { protected String statement(String... parts) { return String.join("?", parts); } + + @Override + protected boolean supportsSavepoints() { + return true; + } + + } diff --git a/vertx-oracle-client/src/main/java/io/vertx/oracleclient/impl/OracleJdbcConnection.java b/vertx-oracle-client/src/main/java/io/vertx/oracleclient/impl/OracleJdbcConnection.java index e0790416b..a77e6494d 100644 --- a/vertx-oracle-client/src/main/java/io/vertx/oracleclient/impl/OracleJdbcConnection.java +++ b/vertx-oracle-client/src/main/java/io/vertx/oracleclient/impl/OracleJdbcConnection.java @@ -206,6 +206,8 @@ private OracleCommand wrap(CommandBase cmd) { action = forExtendedQuery((ExtendedQueryCommand) cmd); } else if (cmd instanceof TxCommand) { action = OracleTransactionCommand.create(connection, context, ((TxCommand) cmd)); + } else if (cmd instanceof SavepointCommand) { + action = OracleSavepointCommand.create(connection, context, ((SavepointCommand) cmd)); } else if (cmd instanceof CloseStatementCommand) { action = new OracleCloseStatementCommand(connection, context); } else if (cmd instanceof CloseCursorCommand) { diff --git a/vertx-oracle-client/src/main/java/io/vertx/oracleclient/impl/commands/OracleSavepointCommand.java b/vertx-oracle-client/src/main/java/io/vertx/oracleclient/impl/commands/OracleSavepointCommand.java new file mode 100644 index 000000000..da9c0edaa --- /dev/null +++ b/vertx-oracle-client/src/main/java/io/vertx/oracleclient/impl/commands/OracleSavepointCommand.java @@ -0,0 +1,52 @@ +/* + * Copyright (c) 2011-2025 Contributors to the Eclipse Foundation + * + * This program and the accompanying materials are made available under the + * terms of the Eclipse Public License 2.0 which is available at + * http://www.eclipse.org/legal/epl-2.0, or the Apache License, Version 2.0 + * which is available at https://www.apache.org/licenses/LICENSE-2.0. + * + * SPDX-License-Identifier: EPL-2.0 OR Apache-2.0 + */ +package io.vertx.oracleclient.impl.commands; + +import io.vertx.core.Future; +import io.vertx.core.internal.ContextInternal; +import io.vertx.sqlclient.spi.protocol.SavepointCommand; +import oracle.jdbc.OracleConnection; + +import java.sql.Statement; + +/** + * Runs a savepoint statement on the JDBC connection. + * + *

Oracle names savepoints with the standard syntax, but has no statement that + * releases one, so {@link SavepointCommand.Kind#RELEASE} never reaches this command. + */ +public class OracleSavepointCommand extends OracleCommand { + + private final SavepointCommand op; + + private OracleSavepointCommand(OracleConnection oracleConnection, ContextInternal connectionContext, SavepointCommand op) { + super(oracleConnection, connectionContext); + this.op = op; + } + + public static OracleSavepointCommand create(OracleConnection oracleConnection, ContextInternal connectionContext, SavepointCommand cmd) { + return new OracleSavepointCommand<>(oracleConnection, connectionContext, cmd); + } + + @Override + protected Future execute() { + if (op.kind() == SavepointCommand.Kind.RELEASE) { + return connectionContext.failedFuture(new UnsupportedOperationException( + "Releasing a savepoint is not supported by Oracle")); + } + String sql = op.sql(); + return executeBlocking(() -> { + try (Statement statement = oracleConnection.createStatement()) { + statement.execute(sql); + } + }).map(op.result()); + } +} diff --git a/vertx-oracle-client/src/main/java/io/vertx/oracleclient/spi/OracleDriver.java b/vertx-oracle-client/src/main/java/io/vertx/oracleclient/spi/OracleDriver.java index d7494ec48..6c7541f09 100644 --- a/vertx-oracle-client/src/main/java/io/vertx/oracleclient/spi/OracleDriver.java +++ b/vertx-oracle-client/src/main/java/io/vertx/oracleclient/spi/OracleDriver.java @@ -62,6 +62,19 @@ public ConnectionFactory createConnectionFactory(Vertx ver return new OracleConnectionFactory(); } + @Override + public boolean supportsSavepoints() { + return true; + } + + /** + * Oracle has no statement that discards a savepoint without rolling back to it. + */ + @Override + public boolean supportsSavepointRelease() { + return false; + } + @Override public SqlConnectionInternal wrapConnection(ContextInternal context, ConnectionFactory factory, Connection connection) { return new OracleConnectionImpl(context, factory, connection); diff --git a/vertx-oracle-client/src/test/java/tests/oracleclient/tck/OracleTransactionTest.java b/vertx-oracle-client/src/test/java/tests/oracleclient/tck/OracleTransactionTest.java index 9f089d355..a5f5cdca1 100644 --- a/vertx-oracle-client/src/test/java/tests/oracleclient/tck/OracleTransactionTest.java +++ b/vertx-oracle-client/src/test/java/tests/oracleclient/tck/OracleTransactionTest.java @@ -11,6 +11,7 @@ package tests.oracleclient.tck; import io.vertx.ext.unit.Async; +import io.vertx.sqlclient.SqlConnection; import io.vertx.ext.unit.TestContext; import io.vertx.ext.unit.junit.VertxUnitRunner; import io.vertx.oracleclient.OracleBuilder; @@ -81,4 +82,36 @@ public void testConstraintViolationIsReported(TestContext ctx) { return conn.preparedQuery(sql, new OraclePrepareOptions().setAutoGeneratedKeys(true)).execute(); }).onComplete(ctx.asyncAssertFailure()); } + + @Override + protected boolean supportsSavepoints() { + return true; + } + + @Override + protected boolean supportsSavepointRelease() { + return false; + } + + /** + * OracleTransactionCommand turns autocommit off to begin and back on when the + * transaction ends. A savepoint runs its own statement on the same JDBC connection, so + * check it leaves that handling alone: the transaction still commits the right rows and + * the connection goes back to the pool once it has. + */ + @Test + public void testSavepointLeavesAutoCommitHandlingIntact(TestContext ctx) { + Async async = ctx.async(); + connector.accept(ctx.asyncAssertSuccess(res -> { + insertMutable(res.client, 1, "before") + .compose(v -> res.tx.createSavepoint()) + .compose(sp -> insertMutable(res.client, 2, "rolled-back").compose(v -> sp.rollback())) + .compose(v -> insertMutable(res.client, 3, "after")) + .compose(v -> res.tx.commit()) + // the connection went back to the pool, so the single pooled connection is free again + .compose(v -> getPool().getConnection().compose(SqlConnection::close)) + .compose(v -> assertMutableIds(ctx, 1, 3)) + .onComplete(ctx.asyncAssertSuccess(v -> async.complete())); + })); + } } diff --git a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/PgConnectionImpl.java b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/PgConnectionImpl.java index d81dbdd02..4273efc7b 100644 --- a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/PgConnectionImpl.java +++ b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/PgConnectionImpl.java @@ -23,7 +23,7 @@ import io.vertx.pgclient.PgNotice; import io.vertx.pgclient.PgNotification; import io.vertx.pgclient.impl.codec.NoticeResponse; -import io.vertx.pgclient.impl.codec.TxFailedEvent; +import io.vertx.pgclient.impl.codec.TxStatusEvent; import io.vertx.pgclient.spi.PgDriver; import io.vertx.sqlclient.codec.SocketConnectionBase; import io.vertx.sqlclient.internal.SqlConnectionBase; @@ -99,9 +99,9 @@ public void handleEvent(Object event) { } else { notice.log(SocketConnectionBase.logger); } - } else if (event instanceof TxFailedEvent) { + } else if (event instanceof TxStatusEvent) { if (tx != null) { - tx.fail(); + tx.status(((TxStatusEvent) event).status()); } } } diff --git a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/PgSocketConnection.java b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/PgSocketConnection.java index 24f9e3edf..ef6c518e3 100644 --- a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/PgSocketConnection.java +++ b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/PgSocketConnection.java @@ -32,7 +32,7 @@ import io.vertx.pgclient.impl.codec.NoticeResponse; import io.vertx.pgclient.impl.codec.PgCodec; import io.vertx.pgclient.impl.codec.PgCommandMessage; -import io.vertx.pgclient.impl.codec.TxFailedEvent; +import io.vertx.pgclient.impl.codec.TxStatusEvent; import io.vertx.sqlclient.codec.CommandMessage; import io.vertx.sqlclient.codec.SocketConnectionBase; import io.vertx.sqlclient.spi.connection.Connection; @@ -42,6 +42,7 @@ import io.vertx.sqlclient.spi.protocol.CommandBase; import io.vertx.sqlclient.spi.protocol.ExtendedQueryCommand; import io.vertx.sqlclient.spi.protocol.InitCommand; +import io.vertx.sqlclient.spi.protocol.SavepointCommand; import io.vertx.sqlclient.spi.protocol.SimpleQueryCommand; import io.vertx.sqlclient.spi.protocol.TxCommand; @@ -117,7 +118,7 @@ Future sendCancelRequestMessage(int processId, int secretKey) { @Override protected void handleMessage(Object msg) { super.handleMessage(msg); - if (msg instanceof Notification || msg instanceof TxFailedEvent || msg instanceof NoticeResponse) { + if (msg instanceof Notification || msg instanceof TxStatusEvent || msg instanceof NoticeResponse) { handleEvent(msg); } } @@ -168,6 +169,15 @@ protected void doSchedule(CommandBase cmd, Completable handler) { SocketConnectionBase.NULL_COLLECTOR, QueryResultHandler.NOOP_HANDLER); super.doSchedule(cmd2, (res, err) -> handler.complete(tx.result(), err)); + } else if (cmd instanceof SavepointCommand) { + SavepointCommand savepoint = (SavepointCommand) cmd; + SimpleQueryCommand cmd2 = new SimpleQueryCommand<>( + savepoint.sql(), + false, + false, + SocketConnectionBase.NULL_COLLECTOR, + QueryResultHandler.NOOP_HANDLER); + super.doSchedule(cmd2, (res, err) -> handler.complete(savepoint.result(), err)); } else { super.doSchedule(cmd, handler); } diff --git a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/PgDecoder.java b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/PgDecoder.java index da7dd7b2d..f1666ea8d 100644 --- a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/PgDecoder.java +++ b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/PgDecoder.java @@ -236,12 +236,11 @@ private void decodeRowDescription(ByteBuf in) { private void decodeReadyForQuery(ChannelHandlerContext ctx, ByteBuf in) { byte id = in.readByte(); if (id == I) { - // IDLE + ctx.fireChannelRead(TxStatusEvent.IDLE); } else if (id == T) { - // ACTIVE + ctx.fireChannelRead(TxStatusEvent.ACTIVE); } else { - // FAILED - ctx.fireChannelRead(TxFailedEvent.INSTANCE); + ctx.fireChannelRead(TxStatusEvent.FAILED); } codec.peek().handleReadyForQuery(); } diff --git a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/TxStatusEvent.java b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/TxStatusEvent.java new file mode 100644 index 000000000..e38543131 --- /dev/null +++ b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/TxStatusEvent.java @@ -0,0 +1,30 @@ +/* + * Copyright (c) 2011-2025 Contributors to the Eclipse Foundation + * + * This program and the accompanying materials are made available under the + * terms of the Eclipse Public License 2.0 which is available at + * http://www.eclipse.org/legal/epl-2.0, or the Apache License, Version 2.0 + * which is available at https://www.apache.org/licenses/LICENSE-2.0. + * + * SPDX-License-Identifier: EPL-2.0 OR Apache-2.0 + */ +package io.vertx.pgclient.impl.codec; + +import io.vertx.sqlclient.impl.TransactionState; + +public class TxStatusEvent { + + public static final TxStatusEvent IDLE = new TxStatusEvent(TransactionState.IDLE); + public static final TxStatusEvent ACTIVE = new TxStatusEvent(TransactionState.ACTIVE); + public static final TxStatusEvent FAILED = new TxStatusEvent(TransactionState.FAILED); + + private final TransactionState status; + + private TxStatusEvent(TransactionState status) { + this.status = status; + } + + public TransactionState status() { + return status; + } +} diff --git a/vertx-pg-client/src/main/java/io/vertx/pgclient/spi/PgDriver.java b/vertx-pg-client/src/main/java/io/vertx/pgclient/spi/PgDriver.java index ed7d00833..3d21357a2 100644 --- a/vertx-pg-client/src/main/java/io/vertx/pgclient/spi/PgDriver.java +++ b/vertx-pg-client/src/main/java/io/vertx/pgclient/spi/PgDriver.java @@ -76,6 +76,11 @@ public int appendQueryPlaceholder(StringBuilder queryBuilder, int index, int cur return index; } + @Override + public boolean supportsSavepoints() { + return true; + } + @Override public SqlConnectionInternal wrapConnection(ContextInternal context, ConnectionFactory factory, Connection connection) { return new PgConnectionImpl((PgConnectionFactory) factory, context, connection); diff --git a/vertx-pg-client/src/test/java/io/vertx/tests/pgclient/tck/PgTransactionTest.java b/vertx-pg-client/src/test/java/io/vertx/tests/pgclient/tck/PgTransactionTest.java index b86ac3558..aa6d1ce99 100644 --- a/vertx-pg-client/src/test/java/io/vertx/tests/pgclient/tck/PgTransactionTest.java +++ b/vertx-pg-client/src/test/java/io/vertx/tests/pgclient/tck/PgTransactionTest.java @@ -10,6 +10,7 @@ */ package io.vertx.tests.pgclient.tck; +import io.vertx.core.Future; import io.vertx.ext.unit.Async; import io.vertx.ext.unit.TestContext; import io.vertx.ext.unit.junit.VertxUnitRunner; @@ -17,12 +18,16 @@ import io.vertx.pgclient.PgException; import io.vertx.sqlclient.Pool; import io.vertx.sqlclient.PoolOptions; +import io.vertx.sqlclient.TransactionRollbackException; +import io.vertx.sqlclient.Tuple; import io.vertx.tests.pgclient.junit.ContainerPgRule; import io.vertx.tests.sqlclient.tck.TransactionTestBase; import org.junit.ClassRule; import org.junit.Test; import org.junit.runner.RunWith; +import java.util.Arrays; + @RunWith(VertxUnitRunner.class) public class PgTransactionTest extends TransactionTestBase { @@ -108,4 +113,521 @@ public void testLongTransaction(TestContext ctx) { })); })); } + + @Test + public void testRollbackToSavepointRestoresTransaction(TestContext ctx) { + Async async = ctx.async(); + connector.accept(ctx.asyncAssertSuccess(res -> { + insertMutable(res.client, 1, "before") + .compose(v -> res.tx.createSavepoint()) + .compose(sp -> insertMutable(res.client, 2, "rolled-back") + .compose(v -> insertMutable(res.client, 1, "duplicate")) + .compose(v -> Future.failedFuture("Expected duplicate key failure")) + .recover(err -> { + assertSqlState(ctx, err, "23505"); + return sp.rollback() + .compose(v -> sp.release()) + .compose(v -> insertMutable(res.client, 3, "after")) + .compose(v -> res.tx.commit()); + })) + .compose(v -> assertMutableIds(ctx, 1, 3)) + .onComplete(ctx.asyncAssertSuccess(v -> async.complete())); + })); + } + + @Test + public void testRollbackInnerSavepointKeepsOuterWork(TestContext ctx) { + Async async = ctx.async(); + connector.accept(ctx.asyncAssertSuccess(res -> { + insertMutable(res.client, 1, "base") + .compose(v -> res.tx.createSavepoint()) + .compose(sp1 -> insertMutable(res.client, 2, "outer") + .compose(v -> res.tx.createSavepoint()) + .compose(sp2 -> insertMutable(res.client, 3, "inner") + .compose(v -> sp2.rollback()) + .compose(v -> insertMutable(res.client, 4, "after-inner-rollback")) + .compose(v -> sp1.release()) + .compose(v -> res.tx.commit()))) + .compose(v -> assertMutableIds(ctx, 1, 2, 4)) + .onComplete(ctx.asyncAssertSuccess(v -> async.complete())); + })); + } + + + + + @Test + public void testCommitCleansUpUnreleasedSavepoint(TestContext ctx) { + Async async = ctx.async(); + connector.accept(ctx.asyncAssertSuccess(res -> { + res.tx.createSavepoint() + .compose(sp -> insertMutable(res.client, 1, "unreleased") + .compose(v -> res.tx.commit())) + .compose(v -> assertMutableIds(ctx, 1)) + .onComplete(ctx.asyncAssertSuccess(v -> async.complete())); + })); + } + + @Test + public void testReleaseOuterSavepointInvalidatesInnerSavepoint(TestContext ctx) { + Async async = ctx.async(); + connector.accept(ctx.asyncAssertSuccess(res -> { + insertMutable(res.client, 1, "base") + .compose(v -> res.tx.createSavepoint()) + .compose(sp1 -> insertMutable(res.client, 2, "outer") + .compose(v -> res.tx.createSavepoint()) + .compose(sp2 -> insertMutable(res.client, 3, "inner") + .compose(v -> sp1.release()) + .compose(v -> sp2.rollback()) + .compose(v -> Future.failedFuture("Expected inner savepoint to be invalidated")) + .recover(err -> { + assertSqlState(ctx, err, "3B001"); + return res.tx.commit(); + }))) + .onComplete(ctx.asyncAssertFailure(err -> { + assertTransactionRollback(ctx, err); + assertMutableIds(ctx).onComplete(ctx.asyncAssertSuccess(v -> async.complete())); + })); + })); + } + + @Test + public void testRollbackOuterSavepointInvalidatesInnerSavepointAndCanRecover(TestContext ctx) { + Async async = ctx.async(); + connector.accept(ctx.asyncAssertSuccess(res -> { + insertMutable(res.client, 1, "base") + .compose(v -> res.tx.createSavepoint()) + .compose(sp1 -> insertMutable(res.client, 2, "outer") + .compose(v -> res.tx.createSavepoint()) + .compose(sp2 -> insertMutable(res.client, 3, "inner") + .compose(v -> sp1.rollback()) + .compose(v -> sp2.release()) + .compose(v -> Future.failedFuture("Expected inner savepoint to be invalidated")) + .recover(err -> { + assertSqlState(ctx, err, "3B001"); + return sp1.rollback() + .compose(v -> insertMutable(res.client, 4, "recovered")) + .compose(v -> res.tx.commit()); + }))) + .compose(v -> assertMutableIds(ctx, 1, 4)) + .onComplete(ctx.asyncAssertSuccess(v -> async.complete())); + })); + } + + @Test + public void testRollbackReleasedSavepointFailsButTransactionCanCommit(TestContext ctx) { + Async async = ctx.async(); + connector.accept(ctx.asyncAssertSuccess(res -> { + res.tx.createSavepoint() + .compose(sp -> sp.release() + .compose(v -> sp.rollback()) + .compose(v -> Future.failedFuture("Expected rollback on released savepoint to fail")) + .recover(err -> { + ctx.assertEquals("Savepoint already released", err.getMessage()); + return insertMutable(res.client, 1, "still-usable") + .compose(v -> res.tx.commit()); + })) + .compose(v -> assertMutableIds(ctx, 1)) + .onComplete(ctx.asyncAssertSuccess(v -> async.complete())); + })); + } + + @Test + public void testReleaseReleasedSavepointFailsButTransactionCanCommit(TestContext ctx) { + Async async = ctx.async(); + connector.accept(ctx.asyncAssertSuccess(res -> { + res.tx.createSavepoint() + .compose(sp -> sp.release() + .compose(v -> sp.release()) + .compose(v -> Future.failedFuture("Expected second release to fail")) + .recover(err -> { + ctx.assertEquals("Savepoint already released", err.getMessage()); + return insertMutable(res.client, 1, "still-usable") + .compose(v -> res.tx.commit()); + })) + .compose(v -> assertMutableIds(ctx, 1)) + .onComplete(ctx.asyncAssertSuccess(v -> async.complete())); + })); + } + + @Test + public void testCreateSavepointFailsWhileTransactionIsFailed(TestContext ctx) { + Async async = ctx.async(); + connector.accept(ctx.asyncAssertSuccess(res -> { + res.tx.createSavepoint() + .compose(sp -> insertMutable(res.client, 1, "before-failure") + .compose(v -> insertMutable(res.client, 1, "duplicate")) + .compose(v -> Future.failedFuture("Expected duplicate key failure")) + .recover(err -> { + assertSqlState(ctx, err, "23505"); + return res.tx.createSavepoint() + .compose(v -> Future.failedFuture("Expected create savepoint to fail in failed transaction")) + .recover(err2 -> { + assertSqlState(ctx, err2, "25P02"); + return sp.rollback() + .compose(v -> insertMutable(res.client, 2, "after-recovery")) + .compose(v -> res.tx.commit()); + }); + })) + .compose(v -> assertMutableIds(ctx, 2)) + .onComplete(ctx.asyncAssertSuccess(v -> async.complete())); + })); + } + + @Test + public void testReleaseSavepointFailsWhileTransactionIsFailed(TestContext ctx) { + Async async = ctx.async(); + connector.accept(ctx.asyncAssertSuccess(res -> { + res.tx.createSavepoint() + .compose(sp -> insertMutable(res.client, 1, "before-failure") + .compose(v -> insertMutable(res.client, 1, "duplicate")) + .compose(v -> Future.failedFuture("Expected duplicate key failure")) + .recover(err -> { + assertSqlState(ctx, err, "23505"); + return sp.release() + .compose(v -> Future.failedFuture("Expected release to fail in failed transaction")) + .recover(err2 -> { + assertSqlState(ctx, err2, "25P02"); + return sp.rollback() + .compose(v -> insertMutable(res.client, 2, "after-recovery")) + .compose(v -> res.tx.commit()); + }); + })) + .compose(v -> assertMutableIds(ctx, 2)) + .onComplete(ctx.asyncAssertSuccess(v -> async.complete())); + })); + } + + + @Test + public void testRollbackToSavepointAfterRepeatedFailedTransactionStatusRestoresTransaction(TestContext ctx) { + Async async = ctx.async(); + connector.accept(ctx.asyncAssertSuccess(res -> { + insertMutable(res.client, 1, "before") + .compose(v -> res.tx.createSavepoint()) + .compose(sp -> insertMutable(res.client, 1, "duplicate") + .compose(v -> Future.failedFuture("Expected duplicate key failure")) + .recover(err -> { + assertSqlState(ctx, err, "23505"); + return res.client.query("SELECT 1") + .execute() + .compose(v -> Future.failedFuture("Expected failed transaction error")) + .recover(err2 -> { + assertSqlState(ctx, err2, "25P02"); + return sp.rollback() + .compose(v -> insertMutable(res.client, 2, "after-recovery")) + .compose(v -> res.tx.commit()); + }); + })) + .compose(v -> assertMutableIds(ctx, 1, 2)) + .onComplete(ctx.asyncAssertSuccess(v -> async.complete())); + })); + } + + @Test + public void testRollbackToSavepointAfterPreparedQueryFailureRestoresTransaction(TestContext ctx) { + Async async = ctx.async(); + connector.accept(ctx.asyncAssertSuccess(res -> { + insertMutable(res.client, 1, "before") + .compose(v -> res.tx.createSavepoint()) + .compose(sp -> res.client.preparedQuery("INSERT INTO mutable (id, val) VALUES ($1, $2)") + .execute(Tuple.of(1, "duplicate")) + .compose(v -> Future.failedFuture("Expected duplicate key failure")) + .recover(err -> { + assertSqlState(ctx, err, "23505"); + return sp.rollback() + .compose(v -> insertMutable(res.client, 2, "after-recovery")) + .compose(v -> res.tx.commit()); + })) + .compose(v -> assertMutableIds(ctx, 1, 2)) + .onComplete(ctx.asyncAssertSuccess(v -> async.complete())); + })); + } + + @Test + public void testRollbackToSavepointAfterPreparedBatchFailureRestoresTransaction(TestContext ctx) { + Async async = ctx.async(); + connector.accept(ctx.asyncAssertSuccess(res -> { + insertMutable(res.client, 1, "before") + .compose(v -> res.tx.createSavepoint()) + .compose(sp -> res.client.preparedQuery("INSERT INTO mutable (id, val) VALUES ($1, $2)") + .executeBatch(Arrays.asList( + Tuple.of(2, "batch-before-error"), + Tuple.of(1, "batch-duplicate"), + Tuple.of(3, "batch-after-error") + )) + .compose(v -> Future.failedFuture("Expected duplicate key failure")) + .recover(err -> { + assertSqlState(ctx, err, "23505"); + return sp.rollback() + .compose(v -> insertMutable(res.client, 4, "after-recovery")) + .compose(v -> res.tx.commit()); + })) + .compose(v -> assertMutableIds(ctx, 1, 4)) + .onComplete(ctx.asyncAssertSuccess(v -> async.complete())); + })); + } + + @Test + public void testSavepointCommandAlreadyInProgress(TestContext ctx) { + Async async = ctx.async(); + connector.accept(ctx.asyncAssertSuccess(res -> { + res.tx.createSavepoint().onComplete(ctx.asyncAssertSuccess(sp -> { + Future first = sp.release(); + Future second = sp.release(); + + second.onComplete(ctx.asyncAssertFailure(err -> { + ctx.assertEquals("Savepoint command already in progress", err.getMessage()); + first + .compose(v -> res.tx.commit()) + .compose(v -> assertMutableIds(ctx)) + .onComplete(ctx.asyncAssertSuccess(x -> async.complete())); + })); + })); + })); + } + + @Test + public void testCreateSavepointAfterCommitRequestedFails(TestContext ctx) { + Async async = ctx.async(); + connector.accept(ctx.asyncAssertSuccess(res -> { + res.client.query("SELECT pg_sleep(0.2)") + .execute(); + + Future commit = res.tx.commit(); + + res.tx.createSavepoint() + .onComplete(ctx.asyncAssertFailure(err -> { + ctx.assertEquals("Transaction already completed", err.getMessage()); + commit.onComplete(ctx.asyncAssertSuccess(v -> async.complete())); + })); + })); + } + + @Test + public void testCreateSavepointAfterRollbackRequestedFails(TestContext ctx) { + Async async = ctx.async(); + connector.accept(ctx.asyncAssertSuccess(res -> { + res.client.query("SELECT pg_sleep(0.2)") + .execute(); + + Future rollback = res.tx.rollback(); + + res.tx.createSavepoint() + .onComplete(ctx.asyncAssertFailure(err -> { + ctx.assertEquals("Transaction already completed", err.getMessage()); + rollback.onComplete(ctx.asyncAssertSuccess(v -> async.complete())); + })); + })); + } + + @Test + public void testRollbackSavepointAfterCommitRequestedFails(TestContext ctx) { + Async async = ctx.async(); + connector.accept(ctx.asyncAssertSuccess(res -> { + res.tx.createSavepoint().onComplete(ctx.asyncAssertSuccess(sp -> { + res.client.query("SELECT pg_sleep(0.2)") + .execute(); + + Future commit = res.tx.commit(); + + sp.rollback().onComplete(ctx.asyncAssertFailure(err -> { + ctx.assertEquals("Transaction already completed", err.getMessage()); + commit.onComplete(ctx.asyncAssertSuccess(v -> async.complete())); + })); + })); + })); + } + + @Test + public void testRollbackSavepointAfterRollbackRequestedFails(TestContext ctx) { + Async async = ctx.async(); + connector.accept(ctx.asyncAssertSuccess(res -> { + res.tx.createSavepoint().onComplete(ctx.asyncAssertSuccess(sp -> { + res.client.query("SELECT pg_sleep(0.2)") + .execute(); + + Future rollback = res.tx.rollback(); + + sp.rollback().onComplete(ctx.asyncAssertFailure(err -> { + ctx.assertEquals("Transaction already completed", err.getMessage()); + rollback.onComplete(ctx.asyncAssertSuccess(v -> async.complete())); + })); + })); + })); + } + + @Test + public void testReleaseSavepointAfterCommitRequestedFails(TestContext ctx) { + Async async = ctx.async(); + connector.accept(ctx.asyncAssertSuccess(res -> { + res.tx.createSavepoint().onComplete(ctx.asyncAssertSuccess(sp -> { + res.client.query("SELECT pg_sleep(0.2)") + .execute(); + + Future commit = res.tx.commit(); + + sp.release().onComplete(ctx.asyncAssertFailure(err -> { + ctx.assertEquals("Transaction already completed", err.getMessage()); + commit.onComplete(ctx.asyncAssertSuccess(v -> async.complete())); + })); + })); + })); + } + + @Test + public void testReleaseSavepointAfterRollbackRequestedFails(TestContext ctx) { + Async async = ctx.async(); + connector.accept(ctx.asyncAssertSuccess(res -> { + res.tx.createSavepoint().onComplete(ctx.asyncAssertSuccess(sp -> { + res.client.query("SELECT pg_sleep(0.2)") + .execute(); + + Future rollback = res.tx.rollback(); + + sp.release().onComplete(ctx.asyncAssertFailure(err -> { + ctx.assertEquals("Transaction already completed", err.getMessage()); + rollback.onComplete(ctx.asyncAssertSuccess(v -> async.complete())); + })); + })); + })); + } + + @Test + public void testWholeTransactionRollbackWithSavepoint(TestContext ctx) { + Async async = ctx.async(); + connector.accept(ctx.asyncAssertSuccess(res -> { + res.tx.createSavepoint() + .compose(sp -> insertMutable(res.client, 1, "before-whole-rollback") + .compose(v -> insertMutable(res.client, 2, "still-rolled-back")) + .compose(v -> res.tx.rollback())) + .compose(v -> assertMutableIds(ctx)) + .onComplete(ctx.asyncAssertSuccess(v -> async.complete())); + })); + } + + @Test + public void testCreateSavepointAfterRollbackCompletedFails(TestContext ctx) { + Async async = ctx.async(); + connector.accept(ctx.asyncAssertSuccess(res -> { + res.tx.rollback().onComplete(ctx.asyncAssertSuccess(v -> { + res.tx.createSavepoint().onComplete(ctx.asyncAssertFailure(err -> { + ctx.assertEquals("Transaction already completed", err.getMessage()); + async.complete(); + })); + })); + })); + } + + @Test + public void testCreateSavepointAfterCommitCompletedFails(TestContext ctx) { + Async async = ctx.async(); + connector.accept(ctx.asyncAssertSuccess(res -> { + res.tx.commit().onComplete(ctx.asyncAssertSuccess(v -> { + res.tx.createSavepoint().onComplete(ctx.asyncAssertFailure(err -> { + ctx.assertEquals("Transaction already completed", err.getMessage()); + async.complete(); + })); + })); + })); + } + + @Test + public void testRollbackSavepointAfterCommitCompletedFails(TestContext ctx) { + Async async = ctx.async(); + connector.accept(ctx.asyncAssertSuccess(res -> { + res.tx.createSavepoint().onComplete(ctx.asyncAssertSuccess(sp -> { + res.tx.commit().onComplete(ctx.asyncAssertSuccess(v -> { + sp.rollback().onComplete(ctx.asyncAssertFailure(err -> { + ctx.assertEquals("Transaction already completed", err.getMessage()); + async.complete(); + })); + })); + })); + })); + } + + @Test + public void testRollbackSavepointAfterRollbackCompletedFails(TestContext ctx) { + Async async = ctx.async(); + connector.accept(ctx.asyncAssertSuccess(res -> { + res.tx.createSavepoint().onComplete(ctx.asyncAssertSuccess(sp -> { + res.tx.rollback().onComplete(ctx.asyncAssertSuccess(v -> { + sp.rollback().onComplete(ctx.asyncAssertFailure(err -> { + ctx.assertEquals("Transaction already completed", err.getMessage()); + async.complete(); + })); + })); + })); + })); + } + + @Test + public void testReleaseSavepointAfterCommitCompletedFails(TestContext ctx) { + Async async = ctx.async(); + connector.accept(ctx.asyncAssertSuccess(res -> { + res.tx.createSavepoint().onComplete(ctx.asyncAssertSuccess(sp -> { + res.tx.commit().onComplete(ctx.asyncAssertSuccess(v -> { + sp.release().onComplete(ctx.asyncAssertFailure(err -> { + ctx.assertEquals("Transaction already completed", err.getMessage()); + async.complete(); + })); + })); + })); + })); + } + + @Test + public void testReleaseSavepointAfterRollbackCompletedFails(TestContext ctx) { + Async async = ctx.async(); + connector.accept(ctx.asyncAssertSuccess(res -> { + res.tx.createSavepoint().onComplete(ctx.asyncAssertSuccess(sp -> { + res.tx.rollback().onComplete(ctx.asyncAssertSuccess(v -> { + sp.release().onComplete(ctx.asyncAssertFailure(err -> { + ctx.assertEquals("Transaction already completed", err.getMessage()); + async.complete(); + })); + })); + })); + })); + } + + @Test + public void testReleaseAfterRollbackToSameSavepoint(TestContext ctx) { + Async async = ctx.async(); + connector.accept(ctx.asyncAssertSuccess(res -> { + res.tx.createSavepoint() + .compose(sp -> insertMutable(res.client, 1, "one") + .compose(v -> sp.rollback()) + .compose(v -> insertMutable(res.client, 2, "two")) + .compose(v -> sp.release()) + .compose(v -> res.tx.commit())) + .compose(v -> assertMutableIds(ctx, 2)) + .onComplete(ctx.asyncAssertSuccess(v -> async.complete())); + })); + } + + + + private void assertSqlState(TestContext ctx, Throwable err, String sqlState) { + ctx.assertTrue(err instanceof PgException); + ctx.assertEquals(sqlState, ((PgException) err).getSqlState()); + } + + private void assertTransactionRollback(TestContext ctx, Throwable err) { + ctx.assertTrue(err instanceof TransactionRollbackException); + } + + @Override + protected boolean supportsSavepoints() { + return true; + } + + /** + * PostgreSQL fails the whole transaction when a statement fails. + */ + @Override + protected boolean statementErrorFailsTransaction() { + return true; + } } diff --git a/vertx-sql-client/src/main/java/io/vertx/sqlclient/Savepoint.java b/vertx-sql-client/src/main/java/io/vertx/sqlclient/Savepoint.java new file mode 100644 index 000000000..8296191f5 --- /dev/null +++ b/vertx-sql-client/src/main/java/io/vertx/sqlclient/Savepoint.java @@ -0,0 +1,42 @@ +/* + * Copyright (c) 2011-2025 Contributors to the Eclipse Foundation + * + * This program and the accompanying materials are made available under the + * terms of the Eclipse Public License 2.0 which is available at + * http://www.eclipse.org/legal/epl-2.0, or the Apache License, Version 2.0 + * which is available at https://www.apache.org/licenses/LICENSE-2.0. + * + * SPDX-License-Identifier: EPL-2.0 OR Apache-2.0 + */ +package io.vertx.sqlclient; + +import io.vertx.codegen.annotations.VertxGen; +import io.vertx.core.Future; + +/** + * A savepoint created from a {@link Transaction}. + * + *

A savepoint marks a position inside the current transaction that can later + * be rolled back to, or released when no longer needed. + */ +@VertxGen +public interface Savepoint { + + /** + * Roll back the current transaction to this savepoint. + * + *

The transaction remains active after a successful rollback. + */ + Future rollback(); + + /** + * Release this savepoint. + * + *

After release, this savepoint can no longer be used. + * + *

Fails with {@link UnsupportedOperationException} when the driver creates + * savepoints but has no statement that releases one, such as Microsoft SQL Server + * and Oracle. The savepoint remains usable for a {@link #rollback()} in that case. + */ + Future release(); +} diff --git a/vertx-sql-client/src/main/java/io/vertx/sqlclient/Transaction.java b/vertx-sql-client/src/main/java/io/vertx/sqlclient/Transaction.java index ccd008011..4a4b2c7f7 100644 --- a/vertx-sql-client/src/main/java/io/vertx/sqlclient/Transaction.java +++ b/vertx-sql-client/src/main/java/io/vertx/sqlclient/Transaction.java @@ -25,6 +25,16 @@ @VertxGen public interface Transaction { + /** + * Create a savepoint in this transaction. + * + *

Fails with {@link UnsupportedOperationException} when the driver does not + * support savepoints. + * + * @return a future notified with the created savepoint + */ + Future createSavepoint(); + /** * Commit the current transaction. */ diff --git a/vertx-sql-client/src/main/java/io/vertx/sqlclient/impl/SavepointImpl.java b/vertx-sql-client/src/main/java/io/vertx/sqlclient/impl/SavepointImpl.java new file mode 100644 index 000000000..a7539bf6f --- /dev/null +++ b/vertx-sql-client/src/main/java/io/vertx/sqlclient/impl/SavepointImpl.java @@ -0,0 +1,69 @@ +/* + * Copyright (c) 2011-2025 Contributors to the Eclipse Foundation + * + * This program and the accompanying materials are made available under the + * terms of the Eclipse Public License 2.0 which is available at + * http://www.eclipse.org/legal/epl-2.0, or the Apache License, Version 2.0 + * which is available at https://www.apache.org/licenses/LICENSE-2.0. + * + * SPDX-License-Identifier: EPL-2.0 OR Apache-2.0 + */ +package io.vertx.sqlclient.impl; + +import io.vertx.core.Future; +import io.vertx.sqlclient.Savepoint; + +public class SavepointImpl implements Savepoint { + + private enum State { + ACTIVE, + PENDING, + RELEASED + } + + private final TransactionImpl transaction; + private final String name; + private State state; + + public SavepointImpl(TransactionImpl transaction, String name) { + this.transaction = transaction; + this.name = name; + this.state = State.ACTIVE; + } + + @Override + public Future rollback() { + return execute(false, () -> transaction.rollbackToSavepoint(name)); + } + + @Override + public Future release() { + return execute(true, () -> transaction.releaseSavepoint(name)); + } + + private Future execute(boolean release, Action action) { + synchronized (this) { + if (state == State.RELEASED) { + return transaction.failedFuture("Savepoint already released"); + } + if (state == State.PENDING) { + return transaction.failedFuture("Savepoint command already in progress"); + } + state = State.PENDING; + } + return action.execute().andThen(ar -> { + synchronized (SavepointImpl.this) { + if (ar.succeeded()) { + state = release ? State.RELEASED : State.ACTIVE; + } else { + state = State.ACTIVE; + } + } + }); + } + + @FunctionalInterface + private interface Action { + Future execute(); + } +} diff --git a/vertx-sql-client/src/main/java/io/vertx/sqlclient/impl/TransactionImpl.java b/vertx-sql-client/src/main/java/io/vertx/sqlclient/impl/TransactionImpl.java index 154155ac2..0077fb4e0 100644 --- a/vertx-sql-client/src/main/java/io/vertx/sqlclient/impl/TransactionImpl.java +++ b/vertx-sql-client/src/main/java/io/vertx/sqlclient/impl/TransactionImpl.java @@ -19,46 +19,91 @@ import io.vertx.core.*; import io.vertx.core.internal.ContextInternal; import io.vertx.core.internal.PromiseInternal; +import io.vertx.sqlclient.Savepoint; import io.vertx.sqlclient.Transaction; import io.vertx.sqlclient.TransactionRollbackException; +import io.vertx.sqlclient.spi.Driver; import io.vertx.sqlclient.spi.connection.Connection; import io.vertx.sqlclient.spi.protocol.CommandBase; +import io.vertx.sqlclient.spi.protocol.SavepointCommand; import io.vertx.sqlclient.spi.protocol.TxCommand; public class TransactionImpl implements Transaction { private final ContextInternal context; private final Connection connection; + private final Driver driver; private final Promise completion; private final Handler endHandler; private int pendingQueries; private boolean ended; - private boolean failed; + private boolean rollbackRequested; + private long savepointSeq; private TxCommand endCommand; + private TransactionState state = TransactionState.ACTIVE; - public TransactionImpl(ContextInternal context, Handler endHandler, Connection connection) { + public TransactionImpl(ContextInternal context, Handler endHandler, Connection connection, Driver driver) { this.context = context; this.connection = connection; + this.driver = driver; this.completion = context.promise(); this.endHandler = endHandler; } public Future begin() { - PromiseInternal promise = context.promise(); - TxCommand begin = new TxCommand<>(TxCommand.Kind.BEGIN, this); - scheduleInternal(begin, wrap(begin, promise)); - return promise.future(); + return submit(new TxCommand<>(TxCommand.Kind.BEGIN, this)); + } + + public void status(TransactionState state) { + synchronized (this) { + this.state = state; + } } - public void fail() { - failed = true; + Future failedFuture(String message) { + return context.failedFuture(message); + } + + @Override + public Future createSavepoint() { + if (!driver.supportsSavepoints()) { + return context.failedFuture(new UnsupportedOperationException( + "Savepoints are not supported by this driver")); + } + + String name; + synchronized (this) { + name = "VX_SP_" + (++savepointSeq); + } + SavepointImpl savepoint = new SavepointImpl(this, name); + return submit(new SavepointCommand<>(SavepointCommand.Kind.CREATE, name, savepoint)); + } + + Future rollbackToSavepoint(String name) { + return submit(new SavepointCommand<>(SavepointCommand.Kind.ROLLBACK_TO, name, null)); + } + + Future releaseSavepoint(String name) { + if (!driver.supportsSavepointRelease()) { + return context.failedFuture(new UnsupportedOperationException( + "Releasing a savepoint is not supported by this driver")); + } + return submit(new SavepointCommand<>(SavepointCommand.Kind.RELEASE, name, null)); + } + + private Future submit(CommandBase cmd) { + PromiseInternal promise = context.promise(); + if (!scheduleInternal(cmd, wrap(promise))) { + promise.fail("Transaction already completed"); + } + return promise.future(); } private void execute(CommandBase cmd, Completable handler) { connection.schedule(cmd, handler); } - private Completable wrap(CommandBase cmd, Completable handler) { + private Completable wrap(Completable handler) { return (res, err) -> { synchronized (TransactionImpl.this) { pendingQueries--; @@ -69,7 +114,7 @@ private Completable wrap(CommandBase cmd, Completable handler) { } public void schedule(CommandBase cmd, Completable handler) { - if (!scheduleInternal(cmd, wrap(cmd, handler))) { + if (!scheduleInternal(cmd, wrap(handler))) { handler.fail("Transaction already completed"); } } @@ -92,7 +137,7 @@ private void checkEnd() { if (pendingQueries > 0 || !ended || endCommand != null) { return; } - TxCommand.Kind kind = failed ? TxCommand.Kind.ROLLBACK : TxCommand.Kind.COMMIT; + TxCommand.Kind kind = rollbackRequested || state == TransactionState.FAILED ? TxCommand.Kind.ROLLBACK : TxCommand.Kind.COMMIT; cmd = new TxCommand<>(kind, null); handler = (res, err) -> { if (err == null) { @@ -113,7 +158,7 @@ private Future end(boolean rollback) { return context.failedFuture("Transaction already complete"); } ended = true; - failed |= rollback; + rollbackRequested |= rollback; } checkEnd(); return completion.future(); diff --git a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/TxFailedEvent.java b/vertx-sql-client/src/main/java/io/vertx/sqlclient/impl/TransactionState.java similarity index 68% rename from vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/TxFailedEvent.java rename to vertx-sql-client/src/main/java/io/vertx/sqlclient/impl/TransactionState.java index 283c1cb43..78e459a00 100644 --- a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/TxFailedEvent.java +++ b/vertx-sql-client/src/main/java/io/vertx/sqlclient/impl/TransactionState.java @@ -8,13 +8,10 @@ * * SPDX-License-Identifier: EPL-2.0 OR Apache-2.0 */ -package io.vertx.pgclient.impl.codec; - -/** - * Event to signal a transaction is failed. - */ -public class TxFailedEvent { - - public static final TxFailedEvent INSTANCE = new TxFailedEvent(); +package io.vertx.sqlclient.impl; +public enum TransactionState { + IDLE, + ACTIVE, + FAILED } diff --git a/vertx-sql-client/src/main/java/io/vertx/sqlclient/internal/SqlConnectionBase.java b/vertx-sql-client/src/main/java/io/vertx/sqlclient/internal/SqlConnectionBase.java index b68418703..6ac4b11be 100644 --- a/vertx-sql-client/src/main/java/io/vertx/sqlclient/internal/SqlConnectionBase.java +++ b/vertx-sql-client/src/main/java/io/vertx/sqlclient/internal/SqlConnectionBase.java @@ -213,7 +213,7 @@ public Future begin() { if (tx != null) { throw new IllegalStateException(); } - tx = new TransactionImpl(context, v -> tx = null, conn); + tx = new TransactionImpl(context, v -> tx = null, conn, driver()); return tx.begin(); } diff --git a/vertx-sql-client/src/main/java/io/vertx/sqlclient/spi/Driver.java b/vertx-sql-client/src/main/java/io/vertx/sqlclient/spi/Driver.java index d9c13658b..f78dad1e6 100644 --- a/vertx-sql-client/src/main/java/io/vertx/sqlclient/spi/Driver.java +++ b/vertx-sql-client/src/main/java/io/vertx/sqlclient/spi/Driver.java @@ -144,4 +144,26 @@ default int appendQueryPlaceholder(StringBuilder queryBuilder, int index, int cu queryBuilder.append("?"); return current; } + + /** + * @return {@code true} when the driver supports savepoints. + */ + default boolean supportsSavepoints() { + return false; + } + + /** + * Whether a savepoint can be released without rolling back to it. + * + *

Some databases, such as Microsoft SQL Server and Oracle, create savepoints + * but offer no statement to discard one. Releasing a savepoint on those drivers + * fails with an {@link UnsupportedOperationException}. + * + *

Only meaningful when {@link #supportsSavepoints()} returns {@code true}. + * + * @return {@code true} when the driver supports releasing a savepoint. + */ + default boolean supportsSavepointRelease() { + return true; + } } diff --git a/vertx-sql-client/src/main/java/io/vertx/sqlclient/spi/protocol/SavepointCommand.java b/vertx-sql-client/src/main/java/io/vertx/sqlclient/spi/protocol/SavepointCommand.java new file mode 100644 index 000000000..d890d1693 --- /dev/null +++ b/vertx-sql-client/src/main/java/io/vertx/sqlclient/spi/protocol/SavepointCommand.java @@ -0,0 +1,56 @@ +/* + * Copyright (c) 2011-2025 Contributors to the Eclipse Foundation + * + * This program and the accompanying materials are made available under the + * terms of the Eclipse Public License 2.0 which is available at + * http://www.eclipse.org/legal/epl-2.0, or the Apache License, Version 2.0 + * which is available at https://www.apache.org/licenses/LICENSE-2.0. + * + * SPDX-License-Identifier: EPL-2.0 OR Apache-2.0 + */ +package io.vertx.sqlclient.spi.protocol; + +public class SavepointCommand extends CommandBase { + + public enum Kind { + CREATE("SAVEPOINT "), + ROLLBACK_TO("ROLLBACK TO SAVEPOINT "), + RELEASE("RELEASE SAVEPOINT "); + + private final String sqlPrefix; + + Kind(String sqlPrefix) { + this.sqlPrefix = sqlPrefix; + } + + public String sql(String name) { + return sqlPrefix + name; + } + } + + private final Kind kind; + private final String name; + private final R result; + + public SavepointCommand(Kind kind, String name, R result) { + this.kind = kind; + this.name = name; + this.result = result; + } + + public Kind kind() { + return kind; + } + + public String name() { + return name; + } + + public String sql() { + return kind.sql(name); + } + + public R result() { + return result; + } +} diff --git a/vertx-sql-client/src/test/java/io/vertx/tests/sqlclient/tck/TransactionTestBase.java b/vertx-sql-client/src/test/java/io/vertx/tests/sqlclient/tck/TransactionTestBase.java index 886dffe52..d3f6f5f6b 100644 --- a/vertx-sql-client/src/test/java/io/vertx/tests/sqlclient/tck/TransactionTestBase.java +++ b/vertx-sql-client/src/test/java/io/vertx/tests/sqlclient/tck/TransactionTestBase.java @@ -19,6 +19,7 @@ import io.vertx.ext.unit.TestContext; import io.vertx.sqlclient.*; import org.junit.After; +import org.junit.Assume; import org.junit.Before; import org.junit.Test; @@ -412,4 +413,320 @@ public void testWithPropagatableConnectionTransactionRollback(TestContext ctx) { })))); }); } + + // --------------------------------------------------------------------------- + // Savepoints + // + // Only the behaviour every database agrees on lives here. Whether a failed + // statement also fails the surrounding transaction is database specific and is + // covered by the driver test classes. + // --------------------------------------------------------------------------- + + /** + * Overridden by the drivers that implement savepoints. + */ + protected boolean supportsSavepoints() { + return false; + } + + /** + * Overridden by the drivers that create savepoints but cannot release one, + * such as Microsoft SQL Server and Oracle. + */ + protected boolean supportsSavepointRelease() { + return true; + } + + /** + * Whether a failed statement also fails the surrounding transaction. + * + *

PostgreSQL puts the transaction in a failed state, every later statement is + * rejected until the transaction is rolled back or rolled back to a savepoint. + * The other databases roll back the failed statement only and leave the + * transaction usable. + */ + protected boolean statementErrorFailsTransaction() { + return false; + } + + /** + * Overridden by the drivers that drop a savepoint once it has been rolled back to. + * Microsoft SQL Server reports "No transaction or savepoint of that name was found" + * on the second rollback. + */ + protected boolean supportsRepeatedRollbackToSavepoint() { + return true; + } + + private void assumeSavepoints() { + Assume.assumeTrue("driver does not support savepoints", supportsSavepoints()); + } + + private void assumeSavepointRelease() { + assumeSavepoints(); + Assume.assumeTrue("driver cannot release a savepoint", supportsSavepointRelease()); + } + + protected Future> insertMutable(SqlConnection client, int id, String val) { + return client.query("INSERT INTO mutable (id, val) VALUES (" + id + ", '" + val + "')").execute(); + } + + protected Future assertMutableIds(TestContext ctx, int... expectedIds) { + return getPool() + .query("SELECT id FROM mutable ORDER BY id") + .execute() + .map(rows -> { + ctx.assertEquals(expectedIds.length, rows.size()); + int index = 0; + for (Row row : rows) { + ctx.assertEquals(expectedIds[index++], row.getInteger("id").intValue()); + } + return null; + }); + } + + @Test + public void testRollbackToSavepointUndoesWorkAfterIt(TestContext ctx) { + assumeSavepoints(); + Async async = ctx.async(); + connector.accept(ctx.asyncAssertSuccess(res -> { + insertMutable(res.client, 1, "before") + .compose(v -> res.tx.createSavepoint()) + .compose(sp -> insertMutable(res.client, 2, "rolled-back") + .compose(v -> sp.rollback()) + .compose(v -> insertMutable(res.client, 3, "after")) + .compose(v -> res.tx.commit())) + .compose(v -> assertMutableIds(ctx, 1, 3)) + .onComplete(ctx.asyncAssertSuccess(v -> async.complete())); + })); + } + + @Test + public void testRollbackInnerThenOuterSavepointKeepsOnlyWorkBeforeOuter(TestContext ctx) { + assumeSavepoints(); + Async async = ctx.async(); + connector.accept(ctx.asyncAssertSuccess(res -> { + insertMutable(res.client, 1, "before-sp1") + .compose(v -> res.tx.createSavepoint()) + .compose(sp1 -> insertMutable(res.client, 2, "between-sp1-sp2") + .compose(v -> res.tx.createSavepoint()) + .compose(sp2 -> insertMutable(res.client, 3, "after-sp2") + .compose(v -> sp2.rollback()) + .compose(v -> insertMutable(res.client, 4, "after-sp2-rollback")) + .compose(v -> sp1.rollback()) + .compose(v -> insertMutable(res.client, 5, "after-sp1-rollback")) + .compose(v -> res.tx.commit()))) + .compose(v -> assertMutableIds(ctx, 1, 5)) + .onComplete(ctx.asyncAssertSuccess(v -> async.complete())); + })); + } + + @Test + public void testRollbackToSameSavepointTwice(TestContext ctx) { + assumeSavepoints(); + Assume.assumeTrue("driver drops the savepoint after a rollback", supportsRepeatedRollbackToSavepoint()); + Async async = ctx.async(); + connector.accept(ctx.asyncAssertSuccess(res -> { + res.tx.createSavepoint() + .compose(sp -> insertMutable(res.client, 1, "first") + .compose(v -> sp.rollback()) + .compose(v -> insertMutable(res.client, 2, "second")) + .compose(v -> sp.rollback()) + .compose(v -> insertMutable(res.client, 3, "third")) + .compose(v -> res.tx.commit())) + .compose(v -> assertMutableIds(ctx, 3)) + .onComplete(ctx.asyncAssertSuccess(v -> async.complete())); + })); + } + + @Test + public void testCanCreateNewSavepointAfterRollbackToSavepoint(TestContext ctx) { + assumeSavepoints(); + Async async = ctx.async(); + connector.accept(ctx.asyncAssertSuccess(res -> { + res.tx.createSavepoint() + .compose(sp1 -> insertMutable(res.client, 1, "first") + .compose(v -> sp1.rollback()) + .compose(v -> res.tx.createSavepoint()) + .compose(sp2 -> insertMutable(res.client, 2, "second") + .compose(v -> sp2.rollback()) + .compose(v -> insertMutable(res.client, 3, "third")) + .compose(v -> res.tx.commit()))) + .compose(v -> assertMutableIds(ctx, 3)) + .onComplete(ctx.asyncAssertSuccess(v -> async.complete())); + })); + } + + @Test + public void testWholeTransactionRollbackDiscardsSavepointWork(TestContext ctx) { + assumeSavepoints(); + Async async = ctx.async(); + connector.accept(ctx.asyncAssertSuccess(res -> { + insertMutable(res.client, 1, "before") + .compose(v -> res.tx.createSavepoint()) + .compose(sp -> insertMutable(res.client, 2, "after") + .compose(v -> res.tx.rollback())) + .compose(v -> assertMutableIds(ctx)) + .onComplete(ctx.asyncAssertSuccess(v -> async.complete())); + })); + } + + @Test + public void testReleaseSavepointKeepsWork(TestContext ctx) { + assumeSavepointRelease(); + Async async = ctx.async(); + connector.accept(ctx.asyncAssertSuccess(res -> { + res.tx.createSavepoint() + .compose(sp -> insertMutable(res.client, 1, "released-scope") + .compose(v -> sp.release()) + .compose(v -> res.tx.commit())) + .compose(v -> assertMutableIds(ctx, 1)) + .onComplete(ctx.asyncAssertSuccess(v -> async.complete())); + })); + } + + @Test + public void testReleaseReleasedSavepointFails(TestContext ctx) { + assumeSavepointRelease(); + Async async = ctx.async(); + connector.accept(ctx.asyncAssertSuccess(res -> { + res.tx.createSavepoint() + .compose(sp -> sp.release().compose(v -> sp.release())) + .onComplete(ctx.asyncAssertFailure(err -> { + res.tx.commit().onComplete(ctx.asyncAssertSuccess(v -> async.complete())); + })); + })); + } + + /** + * A driver that cannot release a savepoint rejects the call rather than pretending + * it worked, and the savepoint stays usable for a rollback. + */ + @Test + public void testReleaseIsRejectedWhenUnsupported(TestContext ctx) { + assumeSavepoints(); + Assume.assumeFalse("driver releases savepoints", supportsSavepointRelease()); + Async async = ctx.async(); + connector.accept(ctx.asyncAssertSuccess(res -> { + res.tx.createSavepoint() + .compose(sp -> insertMutable(res.client, 1, "kept") + .compose(v -> sp.release()) + .transform(ar -> { + ctx.assertTrue(ar.failed(), "release should have been rejected"); + ctx.assertTrue(ar.cause() instanceof UnsupportedOperationException, + "expected an UnsupportedOperationException but got " + ar.cause()); + return insertMutable(res.client, 2, "also-kept").compose(v -> sp.rollback()); + }) + .compose(v -> insertMutable(res.client, 3, "after-rollback")) + .compose(v -> res.tx.commit())) + .compose(v -> assertMutableIds(ctx, 3)) + .onComplete(ctx.asyncAssertSuccess(v -> async.complete())); + })); + } + + @Test + public void testCreateSavepointIsRejectedWhenUnsupported(TestContext ctx) { + Assume.assumeFalse("driver supports savepoints", supportsSavepoints()); + Async async = ctx.async(); + connector.accept(ctx.asyncAssertSuccess(res -> { + res.tx.createSavepoint() + .onComplete(ctx.asyncAssertFailure(err -> { + ctx.assertTrue(err instanceof UnsupportedOperationException, + "expected an UnsupportedOperationException but got " + err); + async.complete(); + })); + })); + } + + @Test + public void testCreateSavepointAfterCommitFails(TestContext ctx) { + assumeSavepoints(); + Async async = ctx.async(); + connector.accept(ctx.asyncAssertSuccess(res -> { + res.tx.commit() + .compose(v -> res.tx.createSavepoint()) + .onComplete(ctx.asyncAssertFailure(err -> async.complete())); + })); + } + + @Test + public void testRollbackSavepointAfterCommitFails(TestContext ctx) { + assumeSavepoints(); + Async async = ctx.async(); + connector.accept(ctx.asyncAssertSuccess(res -> { + res.tx.createSavepoint() + .compose(sp -> res.tx.commit().compose(v -> sp.rollback())) + .onComplete(ctx.asyncAssertFailure(err -> async.complete())); + })); + } + + /** + * A statement that fails rolls back that statement only, the transaction carries on + * and the work around the failure is committed. + */ + @Test + public void testStatementErrorLeavesTransactionUsable(TestContext ctx) { + Assume.assumeFalse("driver fails the transaction", statementErrorFailsTransaction()); + Async async = ctx.async(); + connector.accept(ctx.asyncAssertSuccess(res -> { + insertMutable(res.client, 1, "before") + .compose(v -> insertMutable(res.client, 1, "duplicate")) + .transform(ar -> { + ctx.assertTrue(ar.failed(), "the duplicate key should have failed"); + return insertMutable(res.client, 2, "after"); + }) + .compose(v -> res.tx.commit()) + .compose(v -> assertMutableIds(ctx, 1, 2)) + .onComplete(ctx.asyncAssertSuccess(v -> async.complete())); + })); + } + + /** + * The counterpart: a failed statement leaves the transaction unusable, so the next + * statement is rejected too and nothing is committed. + */ + @Test + public void testStatementErrorFailsTransaction(TestContext ctx) { + Assume.assumeTrue("driver keeps the transaction usable", statementErrorFailsTransaction()); + Async async = ctx.async(); + connector.accept(ctx.asyncAssertSuccess(res -> { + insertMutable(res.client, 1, "before") + .compose(v -> insertMutable(res.client, 1, "duplicate")) + .transform(ar -> { + ctx.assertTrue(ar.failed(), "the duplicate key should have failed"); + return insertMutable(res.client, 2, "after"); + }) + .transform(ar -> { + ctx.assertTrue(ar.failed(), "the transaction should have rejected the next statement"); + return res.tx.rollback(); + }) + .compose(v -> assertMutableIds(ctx)) + .onComplete(ctx.asyncAssertSuccess(v -> async.complete())); + })); + } + + /** + * Rolling back to a savepoint after a failed statement discards the work that followed + * the savepoint and lets the transaction commit, whichever of the two behaviours above + * the database has. + */ + @Test + public void testRollbackToSavepointAfterStatementError(TestContext ctx) { + assumeSavepoints(); + Async async = ctx.async(); + connector.accept(ctx.asyncAssertSuccess(res -> { + insertMutable(res.client, 1, "before") + .compose(v -> res.tx.createSavepoint()) + .compose(sp -> insertMutable(res.client, 2, "rolled-back") + .compose(v -> insertMutable(res.client, 1, "duplicate")) + .transform(ar -> { + ctx.assertTrue(ar.failed(), "the duplicate key should have failed"); + return sp.rollback(); + }) + .compose(v -> insertMutable(res.client, 3, "after")) + .compose(v -> res.tx.commit())) + .compose(v -> assertMutableIds(ctx, 1, 3)) + .onComplete(ctx.asyncAssertSuccess(v -> async.complete())); + })); + } }