spring-projects / spring-projects/spring-framework

Missing transaction cleanup after a reactive publisher is canceled

Open
#36,344 0 comments 0 reactions 0 assignees View on GitHub

Nobody has claimed this yet.

status: waiting-for-triage
Dominant language
Java
Stars
60.2k
Forks
38.8k
Avg merge
5d 2h
Merged PRs (30d)
27

Description

Bug Report

If a reactive publisher with a transactional operator is canceled quickly after being scheduled, the transaction is sometimes not properly cleaned up. If a connection pool is used, the transaction can leak into other requests in this case.

Consider the following reproduction test case (from the repro project below):

Disposable runInTransaction() {
    return Mono.never()
        .as(TransactionalOperator.create(transactionManager)::transactional)
        .subscribe();
}

void checkTransactionLeak() {
    var inTransaction = db.sql("SELECT now() != statement_timestamp() AS in_transaction").fetch().first()
        .map(row -> (boolean) row.get("in_transaction"))
        .blockOptional().orElseThrow();

    assertFalse(inTransaction);
}

@Test
@DirtiesContext(methodMode = DirtiesContext.MethodMode.BEFORE_METHOD)
void immediateCancel() throws InterruptedException { // succeeds
    var disposable = runInTransaction();

    disposable.dispose();

    Thread.sleep(500);
    checkTransactionLeak();
}

@Test
@DirtiesContext(methodMode = DirtiesContext.MethodMode.BEFORE_METHOD)
void brieflyDelayedCancel() throws InterruptedException { // fails
    var disposable = runInTransaction();

    Thread.sleep(1);
    disposable.dispose();

    Thread.sleep(500);
    checkTransactionLeak();
}

@Test
@DirtiesContext(methodMode = DirtiesContext.MethodMode.BEFORE_METHOD)
void longDelayedCancel() throws InterruptedException { // succeeds
    var disposable = runInTransaction();

    Thread.sleep(500);
    disposable.dispose();

    Thread.sleep(500);
    checkTransactionLeak();
}

Canceling the disposable returned from runInTransaction() either immediately or after a longer delay (500ms here) is fine. But canceling it after only a brief delay (1ms here) can cause the transaction to be started but never cleaned up (neither commit nor rollback). Following requests that reuse the same connection will then be erroneously executed within this transaction. This problem can be observed in brieflyDelayedCancel(): the query in checkTransactionLeak() returns true and the log also confirms that the query was executed within a transaction:

Log excerpt 
2026-02-17T15:48:53.514+01:00 DEBUG 112593 --- [transaction-repro] [           main] o.s.r.c.R2dbcTransactionManager          : Creating new transaction with name [null]: org.springframework.transaction.StaticTransactionDefinition@544300a6
2026-02-17T15:48:53.515+01:00 DEBUG 112593 --- [transaction-repro] [tor-tcp-epoll-3] io.r2dbc.pool.ConnectionPool             : Obtaining new connection from the pool
2026-02-17T15:48:53.516+01:00 DEBUG 112593 --- [transaction-repro] [tor-tcp-epoll-3] o.s.r.c.R2dbcTransactionManager          : Acquired Connection [PooledConnection[PostgresqlConnection{client=io.r2dbc.postgresql.client.ReactorNettyClient@7aa1d443, codecs=io.r2dbc.postgresql.codec.DefaultCodecs@40c54182}]] for R2DBC transaction
2026-02-17T15:48:53.517+01:00 DEBUG 112593 --- [transaction-repro] [tor-tcp-epoll-3] o.s.r.c.R2dbcTransactionManager          : Starting R2DBC transaction on Connection [PooledConnection[PostgresqlConnection{client=io.r2dbc.postgresql.client.ReactorNettyClient@7aa1d443, codecs=io.r2dbc.postgresql.codec.DefaultCodecs@40c54182}]] using [ExtendedTransactionDefinition [transactionName='null', readOnly=false, isolationLevel=null, lockWaitTimeout=PT0S]]
2026-02-17T15:48:53.519+01:00 DEBUG 112593 --- [transaction-repro] [tor-tcp-epoll-3] i.r.postgresql.util.FluxDiscardOnCancel  : received cancel signal
2026-02-17T15:48:53.519+01:00 DEBUG 112593 --- [transaction-repro] [tor-tcp-epoll-3] io.r2dbc.postgresql.QUERY                : [cid: 0x2][pid: 77] Executing query: BEGIN READ WRITE
2026-02-17T15:48:53.519+01:00 DEBUG 112593 --- [transaction-repro] [tor-tcp-epoll-3] i.r.p.client.ReactorNettyClient          : [cid: 0x2][pid: 77] Request:  Query{query='BEGIN READ WRITE'}
2026-02-17T15:48:53.519+01:00 DEBUG 112593 --- [transaction-repro] [tor-tcp-epoll-3] i.r.p.client.ReactorNettyClient          : [cid: 0x2][pid: 77] Response: CommandComplete{command=BEGIN, rowId=null, rows=null}
2026-02-17T15:48:53.520+01:00 DEBUG 112593 --- [transaction-repro] [tor-tcp-epoll-3] i.r.p.client.ReactorNettyClient          : [cid: 0x2][pid: 77] Response: ReadyForQuery{transactionStatus=TRANSACTION}
2026-02-17T15:48:54.017+01:00 DEBUG 112593 --- [transaction-repro] [tor-tcp-epoll-3] io.r2dbc.pool.ConnectionPool             : Obtaining new connection from the pool
2026-02-17T15:48:54.017+01:00 DEBUG 112593 --- [transaction-repro] [tor-tcp-epoll-3] o.s.r2dbc.core.DefaultDatabaseClient     : Executing SQL statement [SELECT now() != statement_timestamp() AS in_transaction]
2026-02-17T15:48:54.017+01:00 DEBUG 112593 --- [transaction-repro] [tor-tcp-epoll-3] io.r2dbc.postgresql.QUERY                : [cid: 0x2][pid: 77] Executing query: SELECT now() != statement_timestamp() AS in_transaction
2026-02-17T15:48:54.017+01:00 DEBUG 112593 --- [transaction-repro] [tor-tcp-epoll-3] i.r.p.client.ReactorNettyClient          : [cid: 0x2][pid: 77] Request:  Query{query='SELECT now() != statement_timestamp() AS in_transaction'}
2026-02-17T15:48:54.018+01:00 DEBUG 112593 --- [transaction-repro] [tor-tcp-epoll-3] i.r.p.client.ReactorNettyClient          : [cid: 0x2][pid: 77] Response: RowDescription{fields=[Field{column=0, dataType=16, dataTypeModifier=-1, dataTypeSize=1, format=FORMAT_TEXT, name='in_transaction', table=0}]}
2026-02-17T15:48:54.018+01:00 DEBUG 112593 --- [transaction-repro] [tor-tcp-epoll-3] i.r.p.client.ReactorNettyClient          : [cid: 0x2][pid: 77] Response: DataRow{columns=[UnpooledSlicedByteBuf(ridx: 0, widx: 1, cap: 1/1, unwrapped: AdaptivePoolingAllocator$AdaptiveByteBuf(ridx: 52, widx: 72, cap: 512))]}
2026-02-17T15:48:54.018+01:00 DEBUG 112593 --- [transaction-repro] [tor-tcp-epoll-3] io.r2dbc.pool.PooledConnection           : Releasing connection
2026-02-17T15:48:54.018+01:00 DEBUG 112593 --- [transaction-repro] [tor-tcp-epoll-3] i.r.postgresql.util.FluxDiscardOnCancel  : received cancel signal
2026-02-17T15:48:54.018+01:00 DEBUG 112593 --- [transaction-repro] [tor-tcp-epoll-3] i.r.p.client.ReactorNettyClient          : [cid: 0x2][pid: 77] Response: CommandComplete{command=SELECT, rowId=null, rows=1}
2026-02-17T15:48:54.018+01:00 DEBUG 112593 --- [transaction-repro] [tor-tcp-epoll-3] i.r.p.client.ReactorNettyClient          : [cid: 0x2][pid: 77] Response: ReadyForQuery{transactionStatus=TRANSACTION}

I believe the problem occurs because the Flux.usingWhen() here only executes the asyncCancel handler if the resourceSupplier has already supplied a resource (in this case a transaction). However, if the returned flux is canceled before the resourceSupplier can provide a transaction, the asyncCancel is not called and instead the resourceSupplier is canceled. This is still fine, if the cancellation happens quickly enough and the request has not been sent to the database yet. However, if the request has already been sent, the database will start a transaction that is never cleaned up. Future queries that reuse this existing connection will execute within this transaction.

Reproduction

I was able to reproduce the problem most of the time using this project: transaction-repro.zip

Note that a connection pool with exactly 1 connection is configured to ensure that the connection with the leaked transaction is used for the second query.

Tested Version

Spring Boot 4.0.2

Contributor guide

Open the contributing guide

First steps

  1. Read the whole issue, then the project's contributing guide.
  2. Comment on the issue to say you are picking it up — it saves two people doing the same work.
  3. Fork the repository and make your change on a branch.
  4. Open a pull request that references the issue number.

Research direction

Start at spring-tx/src/main/java/org/springframework/transaction/reactive/TransactionalOperatorImpl.java around the Flux.usingWhen call referenced in the report. Run the reproduction with the one-connection pool, comparing immediateCancel, brieflyDelayedCancel, and longDelayedCancel. Done means cancellation does not leave the reused connection in a transaction and the leak check remains false.

Written by the indexing model from the issue text.

Assessment

Tech stack
java, postgresql, spring, spring-boot
Domain
backend, databases
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Stale
Clarity
Clearly specified
Newbie friendliness
45/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.