[rpc] RequestChannel backpressure transitions can use stale queue state

[fxbing]

Search before asking

• I searched in the issues and found nothing similar.

Fluss version

main (development)

Please describe the bug 🐞

RequestChannel checks the queue threshold before acquiring backpressureLock, but does not recheck it when changing isBackpressureActive.

The queue may change concurrently, causing channels to be paused after the queue has been drained, or resumed after it has reached the backpressure threshold.

The queue size in the log is also read later in the asynchronous EventLoop task, producing misleading messages such as:

Queue size (0) exceeded backpressure threshold (100)

This behavior was introduced by #2065.

Solution

Reconcile the latest queue size and backpressure state under backpressureLock before applying pause or resume.

Capture the queue size at the state transition and use that snapshot in the asynchronous log.

Add concurrent regression tests covering stale pause and resume transitions.

Are you willing to submit a PR?

• I'm willing to submit a PR!

[fxbing]

One possible stale-pause interleaving is:

sequenceDiagram
    participant W as Workers
    participant P as Producer
    participant RC as RequestChannel
    participant EL as EventLoop

    Note over RC: queue=2, state=paused
    W->>RC: W1 polls, queue 2 to 1
    RC->>EL: set resumed and enqueue resume
    Note over RC: W1 is delayed while holding the lock
    P->>RC: put, queue 1 to 2, wait for the lock
    W->>RC: W2 polls twice, queue 2 to 0
    RC-->>W: skip because state is resumed
    P->>RC: acquire lock without rechecking queue
    RC->>EL: set paused and enqueue pause
    EL->>EL: run resume, then pause
    Note over RC,EL: queue=0, autoRead=false

Loading