R2DBC transaction committed after withTransaction signals onComplete

[kstrek]

Hello,

I have spotted a weird behavior related to R2DBC transaction handling in micronaut-data. It seems that in some cases transaction commit can be done after R2dbcOperations#withTransaction (or method annotated with @Transactional) emits the onComplete signal, which obviously doesn't look like a desired behavior.

The easiest way to observe the issue is with additional logs in the micronaut-data code. Here is the commit in my fork with all the necessary changes and a test to showcase the issue: kstrek@244432f

Basically, the test is a simplest possible db insert wrapped in a transaction (it's based on the code found in BookControllerTest.java):

    @Test
    void reproduce() {
        Disposable disposable =
                Mono.from(operations.withTransaction(status ->
                        Mono.from(authorRepository.save(new Author("Stephen King")))
                )).doOnEach(signal -> {
                    if (signal.isOnComplete()) {
                        LOG.info("Transaction completed");
                    }
                }).subscribe();

        Awaitility.await().until(disposable::isDisposed);
    }

Additional logs are introduced in DefaultR2dbcRepositoryOperations after the the commit completes and after the connection is closed.

Logs produced by this test are as follows:

18:28:57.316 [Test worker] DEBUG i.m.d.r.o.DefaultR2dbcRepositoryOperations - Creating a new Connection for DataSource: default
18:28:57.656 [reactor-tcp-epoll-2] DEBUG i.m.d.r.o.DefaultR2dbcRepositoryOperations - Transaction Begin for DataSource: default
18:28:57.725 [reactor-tcp-epoll-2] DEBUG io.micronaut.data.query - Executing SQL Insert: INSERT INTO `author` (`name`) VALUES (?)
18:28:57.735 [reactor-tcp-epoll-2] TRACE io.micronaut.data.query - Binding parameter at position 0 to value Stephen King with data type: STRING
18:28:57.754 [reactor-tcp-epoll-2] DEBUG i.m.d.r.o.DefaultR2dbcRepositoryOperations - Committing transaction for DataSource default.
18:28:57.757 [reactor-tcp-epoll-2] DEBUG i.m.d.r.o.DefaultR2dbcRepositoryOperations - Closing Connection for DataSource: default
18:28:57.760 [reactor-tcp-epoll-2] INFO  example.CommitIssueReproductionTest - Transaction completed
18:28:57.769 [reactor-tcp-epoll-2] DEBUG i.m.d.r.o.DefaultR2dbcRepositoryOperations - [onComplete()] Successfully committed transaction for DataSource default.
18:28:57.775 [reactor-tcp-epoll-2] DEBUG i.m.d.r.o.DefaultR2dbcRepositoryOperations - [onComplete()] Successfully closed connection for DataSource: default

As you see, Transaction completed message is printed before Successfully committed transaction for DataSource default. (and Successfully closed connection for DataSource: default which seems related). Also, another weird thing is that connection is being closed (Closing Connection for DataSource) before commit is finished.

What's interesting, when the publisher inside the withTransaction is an empty mono (i.e. doesn't emit onNext), like this:

Mono.from(authorRepository.save(new Author("Stephen King"))).then()

then everything works as expected:

19:00:56.397 [Test worker] DEBUG i.m.d.r.o.DefaultR2dbcRepositoryOperations - Creating a new Connection for DataSource: default
19:00:56.745 [reactor-tcp-epoll-2] DEBUG i.m.d.r.o.DefaultR2dbcRepositoryOperations - Transaction Begin for DataSource: default
19:00:56.813 [reactor-tcp-epoll-2] DEBUG io.micronaut.data.query - Executing SQL Insert: INSERT INTO `author` (`name`) VALUES (?)
19:00:56.823 [reactor-tcp-epoll-2] TRACE io.micronaut.data.query - Binding parameter at position 0 to value Stephen King with data type: STRING
19:00:56.842 [reactor-tcp-epoll-2] DEBUG i.m.d.r.o.DefaultR2dbcRepositoryOperations - Committing transaction for DataSource default.
19:00:56.866 [reactor-tcp-epoll-2] DEBUG i.m.d.r.o.DefaultR2dbcRepositoryOperations - [onComplete()] Successfully committed transaction for DataSource default.
19:00:56.870 [reactor-tcp-epoll-2] DEBUG i.m.d.r.o.DefaultR2dbcRepositoryOperations - Closing Connection for DataSource: default
19:00:56.879 [reactor-tcp-epoll-2] DEBUG i.m.d.r.o.DefaultR2dbcRepositoryOperations - [onComplete()] Successfully closed connection for DataSource: default
19:00:56.879 [reactor-tcp-epoll-2] INFO  example.CommitIssueReproductionTest - Transaction completed

Am I doing something wrong here? Should I report a bug? I can't find any docs suggesting that publisher wrapped with withTransaction is required to return empty mono. I tried to debug the issue but it seems to involve some weird cancellation logic, which is slightly beyond my comprehension of reactor 😉 . Any help appreciated.

[dstepanov]

Not sure exactly why something like this is happening but you should avoid the operators that are canceling the upstream when using R2DBC, instead of using Mono#from you should use Mono#fromDirect

I have reported some of the behavior that I found weird when using R2DBC pgjdbc/r2dbc-postgresql#428

Maybe there is a way to "fix it" on our side, PRs are welcome 😄

[kstrek]

Using Mono#fromDirect doesn't help, the behavior is exactly the same as with Mono#from. Also, just for the record, the same problem happens when I'm using ReactorCrudRepository instead of ReactiveStreamsCrudRepository (i.e. without any manual wrapping of Publisher into Mono).

Should I convert this discussion to a bug/issue? There is a chance I will find some time to investigate and attempt to fix it but I cannot promise anything 😉.

[dstepanov]

Sure, if you find a way to fix it we can convert it to an issue. Thanks!