Repository: apache/dubbo
File: dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/stream/AbstractTripleClientStream.java (Java)

Code before:
    }

    /**
     * Called when the channel writability changes.
     * This method should be invoked by the transport handler when channelWritabilityChanged is triggered.
     * It synchronously notifies the listener (TripleClientCall) which is responsible for
     * asynchronously triggering all necessary callbacks through its executor.
     */
    protected void onWritabilityChanged() {
        notifyOnReady(false);
    }

    /**
     * Called by InitOnReadyQueueCommand to trigger the initial onReady notification.
     */
    public void triggerInitialOnReady() {
        notifyOnReady(true);
    }

    /**
     * notify listener when stream becomes ready
     *
     * @param forceNotify if true, always trigger onReady (for initial notification);
     *                    if false, only trigger when state changes from "not ready" to "ready"
     */
    private void notifyOnReady(boolean forceNotify) {
        boolean wasReady = lastReadyState;
        boolean isNowReady = isReady();
        lastReadyState = isNowReady;

        // Trigger onReady if:
        // 1. forceNotify is true (initial notification, spurious is OK), or
        // 2. state changes from "not ready" to "ready"
        if (forceNotify || (!wasReady && isNowReady)) {
            listener.onReady();
        }
    }

    class ClientTransportListener extends AbstractH2TransportListener implements H2TransportListener {

        private TriRpcStatus transportError;
        private DeCompressor decompressor;
        private boolean headerReceived;
        private Http2Headers trailers;

        void handleH2TransportError(TriRpcStatus status) {
            writeQueue.enqueue(CancelQueueCommand.createCommand(streamChannelFuture, Http2Error.NO_ERROR));
            rst = true;
            finishProcess(status, null, false);
        }


Review comment (bug):
The notifyOnReady method has a race condition in the read-modify-write sequence. Multiple threads could read the same value of lastReadyState, compute isReady(), and then update lastReadyState, leading to lost state transitions or duplicate notifications. Since this method can be called from different threads (the EventLoop thread via onWritabilityChanged, the sendMessage callback thread, and the InitOnReadyQueueCommand), the entire sequence should be synchronized or use atomic compare-and-set operations to ensure thread safety.
```suggestion
    private synchronized void notifyOnReady(boolean forceNotify) {
```

Code after:

    /**
     * Called when the channel writability changes.
     * This method should be invoked by the transport handler when channelWritabilityChanged is triggered.
     * It synchronously notifies the listener (TripleClientCall) which is responsible for
     * asynchronously triggering all necessary callbacks through its executor.
     */
    protected void onWritabilityChanged() {
        notifyOnReady(false);
    }

    /**
     * Called by InitOnReadyQueueCommand to trigger the initial onReady notification.
     */
    public void triggerInitialOnReady() {
        notifyOnReady(true);
    }

    /**
     * notify listener when stream becomes ready
     *
     * @param forceNotify if true, always trigger onReady (for initial notification);
     *                    if false, only trigger when state changes from "not ready" to "ready"
     */
    private synchronized void notifyOnReady(boolean forceNotify) {
        boolean wasReady = lastReadyState;
        boolean isNowReady = isReady();
        lastReadyState = isNowReady;

        // Trigger onReady if:
        // 1. forceNotify is true (initial notification, spurious is OK), or
        // 2. state changes from "not ready" to "ready"
        if (forceNotify || (!wasReady && isNowReady)) {
            listener.onReady();
        }
    }

    class ClientTransportListener extends AbstractH2TransportListener implements H2TransportListener {

        private TriRpcStatus transportError;
        private DeCompressor decompressor;
        private boolean headerReceived;
        private Http2Headers trailers;

        void handleH2TransportError(TriRpcStatus status) {
            writeQueue.enqueue(CancelQueueCommand.createCommand(streamChannelFuture, Http2Error.NO_ERROR));
            rst = true;
            finishProcess(status, null, false);
        }

        void finishProcess(TriRpcStatus status, Http2Headers trailers, boolean isReturnTriException) {