Hanging when using single connection for concurrent query

[vttranlina]

Bug Report

Versions

• Driver: 1.0.4.RELEASE, 1.0.5.RELEASE
• Database: postgres:16.1
• Java: 21, 11
• OS: Linux (KUbuntu 23.10), MacOs 14.4

Current Behavior

When attempting to execute a DELETE statement with a RETURNING clause using Flux.from(con.createStatement(...)), the application hangs after processing a few elements in the Flux.

My statement

Flux.from(con.createStatement("DELETE FROM message_mailbox WHERE mailbox_id = $1 RETURNING message_id")

Table schema

create table message_mailbox
(
    mailbox_id    uuid                  not null,
    message_uid   bigint                not null,
    message_id    uuid                  not null,
    save_date     timestamp(6),
    primary key (mailbox_id, message_uid)
);

Steps to reproduce

• Insert sample data into the message_mailbox table. (50 rows on my local)

Input Code

public Flux<UUID> deleteByMailboxIdAndReturning(PostgresMailboxId mailboxId) {
        AtomicInteger counter = new AtomicInteger(0);
        AtomicInteger doOnNextCounter = new AtomicInteger(0);

        return postgresExecutor.connection()
            .flatMapMany(con -> Flux.from(con.createStatement("DELETE FROM message_mailbox WHERE mailbox_id = $1 RETURNING message_id")
                    .bind(0, mailboxId.asUuid())
                    .execute())
                .flatMap(result -> result.map((row, rowMetadata) -> {
                    String msgIdWasDeleted = row.get(0, String.class);
                    System.out.println("msg was deleted(" + counter.incrementAndGet() + "): " + msgIdWasDeleted);
                    return msgIdWasDeleted;
                }), 5,15))
            .doOnNext(e -> {
                System.out.println("____doOnNext("+doOnNextCounter.incrementAndGet()+"): " + e);
            })
            .map(UUID::fromString);
    }

It hanging
The console output:

msg was deleted(1): 018ecb78-7627-7c44-88b5-7f4ec96c757f
____doOnNext(1): 018ecb78-7627-7c44-88b5-7f4ec96c757f
msg was deleted(2): 018ecb78-76b9-742c-99cd-02a06d448a27
____doOnNext(2): 018ecb78-76b9-742c-99cd-02a06d448a27
msg was deleted(3): 018ecb78-76d2-76bb-bfdc-55a19366dce6
____doOnNext(3): 018ecb78-76d2-76bb-bfdc-55a19366dce6
msg was deleted(4): 018ecb78-76e9-76d2-9084-6b569cfca91e
____doOnNext(4): 018ecb78-76e9-76d2-9084-6b569cfca91e
msg was deleted(5): 018ecb78-76ff-783d-9be7-2617dfdb4b39
msg was deleted(6): 018ecb78-771e-7c34-a883-c53dbca3cdaf
msg was deleted(7): 018ecb78-7735-79f7-9c4a-468a5ba9eff3
msg was deleted(8): 018ecb78-774c-71b9-81cc-80101fc9553a
msg was deleted(9): 018ecb78-7768-7a86-9f58-8bf5c598adf8
msg was deleted(10): 018ecb78-777f-794a-8a5c-3f4c7c42055f
msg was deleted(11): 018ecb78-7798-7dc5-b44f-7e521679c832
msg was deleted(12): 018ecb78-77b1-72f8-a7ca-79ebccda235f
msg was deleted(13): 018ecb78-77c9-7f52-b58a-81397862e0e3
msg was deleted(14): 018ecb78-77de-7f1a-b2eb-0f53f672d913
msg was deleted(15): 018ecb78-77f6-7bcc-8559-1d2d4e9ddbbc

The number of "doOnNext" should be equal to "msg was deleted", but it not

If I changed the Flux to Mono<List> and revert to Flux::fromIterable
It works
The code just add .collectList().flatMapMany(Flux::fromIterable)

public Flux<UUID> deleteByMailboxIdAndReturning(PostgresMailboxId mailboxId) {
        AtomicInteger counter = new AtomicInteger(0);
        AtomicInteger doOnNextCounter = new AtomicInteger(0);

        return postgresExecutor.connection()
            .flatMapMany(con -> Flux.from(con.createStatement("DELETE FROM message_mailbox WHERE mailbox_id = $1 RETURNING message_id")
                    .bind(0, mailboxId.asUuid())
                    .execute())
                .flatMap(result -> result.map((row, rowMetadata) -> {
                    String msgIdWasDeleted = row.get(0, String.class);
                    System.out.println("msg was deleted(" + counter.incrementAndGet() + "): " + msgIdWasDeleted);
                    return msgIdWasDeleted;
                }), 5,15)).collectList().flatMapMany(Flux::fromIterable)
            .doOnNext(e -> {
                System.out.println("____doOnNext("+doOnNextCounter.incrementAndGet()+"): " + e);
            })
            .map(UUID::fromString);
    }

Then the output is

msg was deleted(1): 018ecb7b-dad2-79e5-96ba-4269fb5339f6
msg was deleted(2): 018ecb7b-db69-7d8a-a475-278b2df269f2
...
msg was deleted(49): 018ecb7b-df6c-7a0a-9258-55817f6cc934
msg was deleted(50): 018ecb7b-df81-779c-be86-c7445186aaff
____doOnNext(1): 018ecb7b-dad2-79e5-96ba-4269fb5339f6
____doOnNext(2): 018ecb7b-db69-7d8a-a475-278b2df269f2
...
____doOnNext(49): 018ecb7b-df6c-7a0a-9258-55817f6cc934
____doOnNext(50): 018ecb7b-df81-779c-be86-c7445186aaff

Expected behavior/code

The Flux.from(connection.createStatement(...)) should execute normally without hanging.

Additional context

• The issue occurs not only with DELETE ... RETURNING queries but also with SELECT queries, although the latter is less frequently reproducible.
• No error messages or stack traces are observed during the hang.

---

updated:
The code for reproduce is more complex: #650 (comment)

[vttranlina]

I updated the thread dump when the reactor process hanging
threads_report1.txt

[mp911de]

There's no blocked thread, the reported issue requires a bit more digging. I suggest enabling debug logging for the driver so that you see at which Postgres frame the process gets locked up.

[vttranlina]

There's no blocked thread, the reported issue requires a bit more digging. I suggest enabling debug logging for the driver so that you see at which Postgres frame the process gets locked up.

I created a simple project to reproduce it:
https://github.com/vttranlina/r2dbc-postgresql-test.git

• test class: R2dbcPostgresqlTest

I updated the re-produce code (it is more complex a bit than what I wrote in the description above)

I tried to print all debug logs, but I don't see anything that I can dig more

[vttranlina]

This issue still persists in the latest version of r2dbc-postgresql (1.0.7.RELEASE).

After trying to gain a deeper understanding of the issue, I realized that the hanging does not occur solely because of DELETE RETURNING
Instead, it happens when a connection-A is being used for a query, and during the process of receiving results (not yet completed), the connection-A is used to execute another query. This causes the hang.

In the example I provided, when I created two separate connections, the issue did not occur.

[vttranlina]

I changed the title of ticket, I think it will more clear
Here is new log from re-produce code
r2dbc-postgresql-hanging.log

A lot of log from ReactorNettyClient class, Can any help me?