Transaction isolation level not working with R2dbcTransactionManager

[sasavilic]

Spring boot: 3.1.0 and 3.1.1 (SNAPSHOT)
Database: PostgreSQL

@Service
public class TxTester implements ApplicationRunner {
    private final Logger logger = LoggerFactory.getLogger(TxTester.class);

    private final DatabaseClient databaseClient;
    private final TransactionalOperator rrOperator;
    private final TransactionalOperator nOperator;

    @Autowired
    public TxTester(DatabaseClient databaseClient, ReactiveTransactionManager transactionManager) {
        this.databaseClient = databaseClient;
        this.rrOperator = TransactionalOperator.create(transactionManager, new TransactionDefinition() {
            @Override
            public int getIsolationLevel() {
                return TransactionDefinition.ISOLATION_REPEATABLE_READ;
            }
        });
        this.nOperator = TransactionalOperator.create(transactionManager, new TransactionDefinition() {
            @Override
            public int getPropagationBehavior() {
                return TransactionDefinition.PROPAGATION_REQUIRES_NEW;
            }
        });
    }

    @Override
    public void run(ApplicationArguments args) throws Exception {
        logger.info("Creating table");
        databaseClient.sql("CREATE TABLE IF NOT EXISTS person(id INT PRIMARY KEY, name VARCHAR)").then().block();
        logger.info("Adding sample record");
        databaseClient.sql("INSERT INTO person(id, name) VALUES (1, 'John') ON CONFLICT (id) DO UPDATE SET name = EXCLUDED.name").then().block();

        logger.info("Getting row with select for update");
        var name = databaseClient.sql("SELECT 1")
                .then()
                .doOnSuccess(x -> logger.info("Now updating row in different transaction"))
                .then(databaseClient.sql("UPDATE person SET name = 'Mark' WHERE id = 1").then().as(nOperator::transactional))
                .doOnSuccess(x -> logger.info("Fetching row with FOR UPDATE. Expecting to fail"))
                .then(databaseClient.sql("SELECT * FROM person WHERE id = 1 FOR UPDATE")
                        .map((row, metadata) -> row.get("name", String.class))
                        .one()
                )
                .as(rrOperator::transactional)
                .block();
        logger.info("Got name: {}. This should not happen!", name);
    }
}

Expected is something like (works in spring-boot 2.4.13):

Caused by: org.springframework.dao.ConcurrencyFailureException: executeMany; SQL [SELECT * FROM person WHERE id = 1 FOR UPDATE]; could not serialize access due to concurrent update; nested exception is io.r2dbc.postgresql.ExceptionFactory$PostgresqlRollbackException: [40001] could not serialize access due to concurrent update

I also investigated traffic with wireshark between PostgreSQL and my demo project, there is no isolation level information being set.

[sasavilic]

After deeper investigation, I believe the problem is following:

In R2dbcTransactionManager we have:

@Override
	protected Mono<Void> doBegin(TransactionSynchronizationManager synchronizationManager, Object transaction,
			TransactionDefinition definition) throws TransactionException {

                        ....

			return connectionMono.flatMap(con -> switchAutoCommitIfNecessary(con, transaction)
					.then(Mono.from(doBegin(definition, con)))
					.then(prepareTransactionalConnection(con, definition))
					.doOnSuccess(v -> {
						txObject.getConnectionHolder().setTransactionActive(true);
						Duration timeout = determineTimeout(definition);
						if (!timeout.isNegative() && !timeout.isZero()) {
							txObject.getConnectionHolder().setTimeoutInMillis(timeout.toMillis());
						}
						// Bind the connection holder to the thread.
						if (txObject.isNewConnectionHolder()) {
							synchronizationManager.bindResource(obtainConnectionFactory(), txObject.getConnectionHolder());
						}
					}).thenReturn(con).onErrorResume(e -> {
						if (txObject.isNewConnectionHolder()) {
							return ConnectionFactoryUtils.releaseConnection(con, obtainConnectionFactory())
									.doOnTerminate(() -> txObject.setConnectionHolder(null, false))
									.then(Mono.error(e));
						}
						return Mono.error(e);
					})).onErrorResume(e -> {
						CannotCreateTransactionException ex = new CannotCreateTransactionException(
								"Could not open R2DBC Connection for transaction", e);
						return Mono.error(ex);
					});

switchAutoCommitIfNecessary will call setAutoCommit(false) before calling doBegin(definition, con). setAutoCommit(false) actually starts connection and thus doBegin(definition, con) calling con.beginTransaction(transactionDefinition) is actually no-op.

I believe that calling setAutoCommit is not necessary any more, because based on Connection interface autoCommit is disabled by beginTransaction(definition) method.:

    /**
     * Begins a new transaction.  Calling this method disables {@link #isAutoCommit() auto-commit} mode.  Beginning the transaction may fail if the {@link TransactionDefinition} conflicts with the
     * connection configuration.
     *
     * @param definition attributes for the transaction.
     * @return a {@link Publisher} that indicates that the transaction is open
     * @since 0.9
     */
    Publisher<Void> beginTransaction(TransactionDefinition definition);

To prove my theory, I have created wrapper around ConnectionFactory and basically ignore setAutoCommit call:

@Component
@Aspect
public class FactoryWrapper {
    @Around("execution(* io.r2dbc.spi.ConnectionFactory.create())")
    public Object wrapConnection(ProceedingJoinPoint pjp) throws Throwable {
        Publisher<? extends Connection> publisher = (Publisher<? extends Connection>)pjp.proceed();
        return Mono.from(publisher)
                .map(GuardAutoCommit::new);
    }

    private static class GuardAutoCommit implements Connection, Wrapped<Connection> {
        private final Connection delegate;

        private GuardAutoCommit(Connection delegate) {
            this.delegate = delegate;
        }

        @Override
        public boolean isAutoCommit() {
            return delegate.isAutoCommit();
        }

        @Override
        public Publisher<Void> setAutoCommit(boolean autoCommit) {
            return Mono.empty();
        }

        ....

        @Override
        public Connection unwrap() {
            return delegate;
        }
    }
}

and with that, it works.

[jhoeller]

Thanks for the report! We seem to have missed the implicit auto-commit handling in R2DBC 0.9/1.0 when doing the upgrade.

@mp911de I hope my corresponding changes make sense there. Since there is only cleanup step left, I've folded that logging/completion code into the release connection step itself.