[konlue]

Problem

ShenyuGrpcClient.invoke() sends CLIENT_STREAMING and BIDI_STREAMING requests with an unconditional requestList.forEach(requestObserver::onNext) and never observes ClientCallStreamObserver.isReady(). onNext can enqueue messages before the peer can consume them, creating avoidable outbound buffering when the backend reads slowly.

Change

• Pass a ClientResponseObserver wrapper for the two streaming branches. grpc-java calls beforeStart(ClientCallStreamObserver) and only afterwards freezes the stream, so this is the supported hook to install setOnReadyHandler; no cast to an internal adapter is needed.
• ReadinessAwareRequestSender drains the existing request list with a cursor only while isReady(). A CAS single-drain lock serializes reentrant/concurrent readiness callbacks, and readiness is rechecked after releasing the lock so a callback that races the loop cannot strand the remaining messages.
• onCompleted() is called exactly once after the last message. Termination (response completion, response error, returned-future cancellation) stops further sends.
• A send failure cancels the call via ClientCallStreamObserver.cancel(...), so the response path ends in onError and the returned future cannot hang.
• No new queue or executor: the caller-provided list is the only buffer. Unary and server-streaming paths are unchanged.

Tests

• ShenyuGrpcClientTest: client-streaming and bidi wiring through MockedStatic<ClientCalls> with a non-ready fake observer — no sends while not ready, ordered delivery and one completion after readiness, and cancellation of the returned future stops pending sends.
• ReadinessAwareRequestSenderTest: all-ready; not-ready then ready; mid-drain readiness change; reentrant onReady (no duplicates, order preserved); termination before/after a partial drain; a send failure cancels once and does not complete; an empty list completes once; the readiness handler is installed before the first send.
• Red on master: the streaming readiness test fails because the eager loop sends while the fake is not ready. Green after the change.

Verification

• Targeted tests → 15/15
• shenyu-plugin-grpc full module → 84/84
• checkstyle:check → 0 violations

No slow-reading-backend benchmark was run; deterministic readiness/cancellation tests cover the contract (the issue lists the slow-backend exercise as proposed validation). Unary and server-streaming behavior is unchanged.

Fixes #7927

[Aias00]

Approved. This is a real fix rather than a cosmetic rewrite: the previous code pushed the whole request list through onNext regardless of isReady(), and the new sender gates every send on readiness while preserving order, exactly-once submission and a single termination.

What I verified

• Read the production diff plus both new classes. setOnReadyHandler is installed from beforeStart, and I confirmed against the pinned runtime that this is the only legal window: grpc-java 1.65.1 ClientCalls.startCall constructs StreamObserverToCallListenerAdapter before call.start(...), and that constructor invokes beforeStart(adapter); CallToStreamObserverAdapter.setOnReadyHandler throws IllegalStateException("Cannot alter onReadyHandler after call started. Use ClientResponseObserver") once freeze() runs immediately after beforeStart. The handler is therefore installed before the call starts and is read on every readiness delivery (StreamObserverToCallListenerAdapter.onReady -> adapter.onReadyHandler.run()). So the fix is live, not silently falling back to the legacy loop.
• The explicit drain() in start() covers the "already ready at start" case where no false->true transition would ever deliver onReady. Good call.
• Rollback: reverting only ShenyuGrpcClient.java to base (keeping both new main classes and every test) fails exactly the 3 new streaming tests with expected: <true> but was: <false> ("must not send while the stream is not ready" x2 plus the cancellation case), while the 4 pre-existing tests stay green.
• Mutation on ReadinessAwareRequestSender: dropping observer.isReady() from drainLoop fails 3 tests; making terminate() a no-op fails 3 tests across both classes. Both guards are pinned.
• mvn -pl ...shenyu-plugin-grpc test -> 91/91 green; checkstyle:check clean.
• Concurrency: I walked the CAS/lock/re-check loop for lost wake-ups. A caller that loses the CAS returns, and the holder re-checks terminal / cursor / isReady() after releasing the lock, so a readiness transition observed only by the loser is still picked up. onCompleted is reachable only by the lock holder, so it stays exactly-once.
• No half-open stream on termination: MessageWriter.triggerOverflow cancels the call, and a normal response-side completion already ends both directions.

Non-blocking comments

1. reentrantReadinessCallbackDoesNotDuplicateMessages does not actually pin the CAS single-flight guard (details inline).
2. The two FakeRequestObserver classes duplicate ~100 lines; if extracted, modelling the frozen contract would protect the beforeStart invariant (details inline).
3. Optional: no RELEASE-NOTES.md entry, while other recently merged fixes (#7969, #8043) added one.
4. #7927's own acceptance list also asks for a slow-reading backend run with a large finite request list. The fakes cannot show convergence, so that is worth recording on the issue rather than blocking here.

Nice work on keeping the request list as a snapshot with no extra queue and no extra executor.

[Aias00]

Mutation check: replacing draining.compareAndSet(false, true) with a plain draining.set(true) keeps all 8 tests green. With only 3 messages the reentrant fireReady() unwinds in 3 frames, and cursor.getAndIncrement() still prevents duplicates, so the guard this test is named after is not actually pinned.

The guard's real job is bounding reentrant recursion depth. A case with a few thousand requests and a reentrant fireReady() inside onNext, asserting no StackOverflowError, would catch its removal.

[Aias00]

This fake duplicates most of ReadinessAwareRequestSenderTest.FakeRequestObserver. If the two are extracted into a shared fixture, consider modelling the grpc-java adapter contract as well: CallToStreamObserverAdapter.setOnReadyHandler throws IllegalStateException once freeze() runs right after beforeStart.

Today a refactor that moved setOnReadyHandler out of beforeStart (for example into start()) would still pass both fakes, while production would throw on every client/bidi call. Making the fake throw once started would turn that invariant into a test failure instead of a runtime failure.