[BUG] SentinelGatewayFilter 限流失效，未能阻断流程

[15911056747]

问题：
在springcloudgateway项目中 使用了内置的SentinelGatewayFilter 做限流控制；在reactor-core版本有3.3.10 升级到3.4.34 版本后，限流失效，未能中断流程依然外调了。
原因：
reactor-core 3.3.10版本MonoIgnoreThen.ThenIgnoreMain 的 subscribeNext()方法中先判断了isCancelled()而reactor-core 3.4.34版本MonoIgnoreThen.ThenIgnoreMain 的 subscribeNext()方法中先判断了publisher 是否是Callable类型是Callable类型才会看isCancelled()状态 非Callable的直接向下订阅了
具体流程:
1.在SentinelGatewayFilter订阅了限流 SentinelReactorSubscriber
2.在内置NettyWriteResponseFilter 通过chain.filter(exchange).then(Mono.defer(()-{}))触发SentinelReactorSubscriber
3.在MonoIgnoreThen 中通过actual.onSubscribe()触发执行
4.在内置SentinelReactorSubscriber 中触发限流 设置cancel状态 执行onError()
5.异常执行完成后，在MonoIgnoreThen.ThenIgnoreMain 的 subscribeNext() 中 由于ignoreMono不是Callable类型继续向下订阅了走了后边的filter

[kaori-seasons]

@LearningGp 经过代码走读之后发现 SentinelReactorSubscriber 继承自 InheritableBaseSubscriber，在 onError 方法中通过原子操作设置 subscription 为已取消状态

但在 reactor-core 3.4.34 中，MonoIgnoreThen 的行为变化导致这个取消状态没有被正确检查。

我写了一个复现的测试用例

package com.alibaba.csp.sentinel.adapter.reactor;

import com.alibaba.csp.sentinel.node.ClusterNode;
import com.alibaba.csp.sentinel.slots.block.BlockException;
import com.alibaba.csp.sentinel.slots.block.flow.FlowRule;
import com.alibaba.csp.sentinel.slots.block.flow.FlowRuleManager;
import com.alibaba.csp.sentinel.slots.clusterbuilder.ClusterBuilderSlot;
import org.junit.After;
import org.junit.Test;
import reactor.core.publisher.Mono;
import reactor.test.StepVerifier;

import java.util.ArrayList;
import java.util.Collections;
import java.util.concurrent.atomic.AtomicBoolean;

import static org.junit.Assert.*;

/**
 * 测试用例：复现 reactor-core 3.4.x 版本下 .then() 操作符导致的限流失效问题
 * 
 * 模拟 Spring Cloud Gateway 中 NettyWriteResponseFilter 的使用场景：
 * chain.filter(exchange).then(Mono.defer(() -> writeResponse()))
 */
public class MonoIgnoreThenFlowControlTest {

    @Test
    public void testFlowControlWithThenOperator() {
        String resourceName = "test_then_flow_control";
        AtomicBoolean downstreamExecuted = new AtomicBoolean(false);
        
        // 设置限流规则：QPS = 0，即完全阻断
        FlowRuleManager.loadRules(Collections.singletonList(
            new FlowRule(resourceName).setCount(0)
        ));

        // 模拟 Gateway filter chain 的场景
        Mono<Void> upstreamMono = Mono.empty()
            .transform(new SentinelReactorTransformer<>(resourceName));
        
        // 模拟 NettyWriteResponseFilter 的 .then() 操作
        Mono<Void> downstreamMono = Mono.defer(() -> {
            downstreamExecuted.set(true);
            System.out.println("Downstream executed - This should NOT happen when flow control is triggered!");
            return Mono.empty();
        });
        
        Mono<Void> result = upstreamMono.then(downstreamMono);

        // 验证：应该抛出 BlockException
        StepVerifier.create(result)
            .expectError(BlockException.class)
            .verify();

        // 验证限流统计
        ClusterNode cn = ClusterBuilderSlot.getClusterNode(resourceName);
        assertNotNull(cn);
        assertEquals(0, cn.passQps(), 0.01);
        assertEquals(1, cn.blockRequest());
        
        // 关键验证：下游不应该被执行
        // 在 reactor-core 3.3.x 中此断言会通过
        // 在 reactor-core 3.4.x 中此断言会失败（downstreamExecuted 为 true）
        assertFalse("Downstream should NOT be executed when flow control is triggered", 
                    downstreamExecuted.get());
    }

    @Test
    public void testFlowControlWithMultipleThenChain() {
        String resourceName = "test_multiple_then_flow_control";
        AtomicBoolean downstream1Executed = new AtomicBoolean(false);
        AtomicBoolean downstream2Executed = new AtomicBoolean(false);
        
        FlowRuleManager.loadRules(Collections.singletonList(
            new FlowRule(resourceName).setCount(0)
        ));

        // 模拟多个 filter 的链式调用
        Mono<Void> result = Mono.empty()
            .transform(new SentinelReactorTransformer<>(resourceName))
            .then(Mono.defer(() -> {
                downstream1Executed.set(true);
                return Mono.empty();
            }))
            .then(Mono.defer(() -> {
                downstream2Executed.set(true);
                return Mono.empty();
            }));

        StepVerifier.create(result)
            .expectError(BlockException.class)
            .verify();

        assertFalse("First downstream should NOT be executed", downstream1Executed.get());
        assertFalse("Second downstream should NOT be executed", downstream2Executed.get());
    }

    @Test
    public void testFlowControlWithThenReturnValue() {
        String resourceName = "test_then_return_flow_control";
        AtomicBoolean downstreamExecuted = new AtomicBoolean(false);
        
        FlowRuleManager.loadRules(Collections.singletonList(
            new FlowRule(resourceName).setCount(0)
        ));

        // 测试 thenReturn 场景
        Mono<String> result = Mono.just("upstream")
            .transform(new SentinelReactorTransformer<>(resourceName))
            .then(Mono.defer(() -> {
                downstreamExecuted.set(true);
                return Mono.just("downstream");
            }));

        StepVerifier.create(result)
            .expectError(BlockException.class)
            .verify();

        assertFalse("Downstream should NOT be executed", downstreamExecuted.get());
    }

    @After
    public void cleanup() {
        FlowRuleManager.loadRules(new ArrayList<>());
    }
}

运行此测试用例：

• 在 reactor-core 3.3.x 环境下： 测试通过，下游不会执行
• 在 reactor-core 3.4.x 环境下： 测试失败，下游仍然被执行

可用的解决方案

方案一：使用 onErrorResume 确保异常传播（推荐）

修改 SentinelGatewayFilter 的实现，确保 BlockException 能够正确中断后续流程：

@Override
public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
    Route route = exchange.getAttribute(ServerWebExchangeUtils.GATEWAY_ROUTE_ATTR);

    Mono<Void> asyncResult = chain.filter(exchange);
    if (route != null) {
        String routeId = route.getId();
        Object[] params = paramParser.parseParameterFor(routeId, exchange,
            r -> r.getResourceMode() == SentinelGatewayConstants.RESOURCE_MODE_ROUTE_ID);
        String origin = Optional.ofNullable(GatewayCallbackManager.getRequestOriginParser())
            .map(f -> f.apply(exchange))
            .orElse("");
        
        // 关键改动：使用 compose 替代 transform，并添加显式的错误处理
        asyncResult = asyncResult.compose(mono -> 
            mono.transform(new SentinelReactorTransformer<>(
                new EntryConfig(routeId, ResourceTypeConstants.COMMON_API_GATEWAY,
                    EntryType.IN, 1, params, new ContextConfig(contextName(routeId), origin))))
            // 确保 BlockException 立即传播，阻止后续订阅
            .onErrorResume(BlockException.class, ex -> Mono.error(ex))
        );
    }

    Set<String> matchingApis = pickMatchingApiDefinitions(exchange);
    for (String apiName : matchingApis) {
        Object[] params = paramParser.parseParameterFor(apiName, exchange,
            r -> r.getResourceMode() == SentinelGatewayConstants.RESOURCE_MODE_CUSTOM_API_NAME);
        asyncResult = asyncResult.compose(mono -> 
            mono.transform(new SentinelReactorTransformer<>(
                new EntryConfig(apiName, ResourceTypeConstants.COMMON_API_GATEWAY,
                    EntryType.IN, 1, params)))
            .onErrorResume(BlockException.class, ex -> Mono.error(ex))
        );
    }

    return asyncResult;
}

方案二：自定义 SentinelReactorSubscriber 优化（定制方案）

创建一个增强版的 SentinelReactorSubscriber，在 onError 后主动取消上游订阅：

public class EnhancedSentinelReactorSubscriber<T> extends SentinelReactorSubscriber<T> {
    
    public EnhancedSentinelReactorSubscriber(EntryConfig entryConfig,
                                           CoreSubscriber<? super T> actual,
                                           boolean unary) {
        super(entryConfig, actual, unary);
    }

    @Override
    protected void hookOnError(Throwable t) {
        // 在传播错误之前，确保上游订阅被取消
        if (t instanceof BlockException) {
            Subscription s = upstream();
            if (s != null) {
                s.cancel();
            }
        }
        super.hookOnError(t);
    }
}

方案三：包装 Mono 为 Callable（兼容方案）

利用 reactor-core 3.4.x 对 Callable 类型的特殊处理，将结果包装为 Callable：

@Override
public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
    Route route = exchange.getAttribute(ServerWebExchangeUtils.GATEWAY_ROUTE_ATTR);

    // 使用 fromCallable 包装，确保在新版本 reactor-core 中正确检查 cancel 状态
    Mono<Void> asyncResult = Mono.fromCallable(() -> {
        return null; // Void type
    }).flatMap(v -> chain.filter(exchange));
    
    if (route != null) {
        String routeId = route.getId();
        Object[] params = paramParser.parseParameterFor(routeId, exchange,
            r -> r.getResourceMode() == SentinelGatewayConstants.RESOURCE_MODE_ROUTE_ID);
        String origin = Optional.ofNullable(GatewayCallbackManager.getRequestOriginParser())
            .map(f -> f.apply(exchange))
            .orElse("");
        asyncResult = asyncResult.transform(
            new SentinelReactorTransformer<>(new EntryConfig(routeId, 
                ResourceTypeConstants.COMMON_API_GATEWAY,
                EntryType.IN, 1, params, new ContextConfig(contextName(routeId), origin)))
        );
    }

    // ... 其余代码保持不变
    return asyncResult;
}

方案四：固定 reactor-core 版本（临时方案）

在项目的 pom.xml 中显式指定兼容的 reactor-core 版本：

<properties>
    <reactor-core.version>3.3.22.RELEASE</reactor-core.version>
</properties>

<dependencyManagement>
    <dependencies>
        <dependency>
            <groupId>io.projectreactor</groupId>
            <artifactId>reactor-core</artifactId>
            <version>${reactor-core.version}</version>
        </dependency>
    </dependencies>
</dependencyManagement>

推荐：方案一（使用 onErrorResume）

• 不需要修改 Sentinel 源码
• 兼容所有 reactor-core 版本
• 明确的错误处理语义
• 易于理解和维护

实施步骤：

1. 创建自定义 Gateway Filter 替代直接使用 SentinelGatewayFilter

2. 配置 Gateway

@Configuration
public class GatewayConfiguration {
    
    @Bean
    @Order(Ordered.HIGHEST_PRECEDENCE)
    public GlobalFilter sentinelGatewayFilter() {
        return new EnhancedSentinelGatewayFilter();
    }
}

3. 添加全局异常处理器

@Component
@Order(Ordered.HIGHEST_PRECEDENCE)
public class SentinelBlockExceptionHandler implements WebExceptionHandler {
    
    @Override
    public Mono<Void> handle(ServerWebExchange exchange, Throwable ex) {
        if (ex instanceof BlockException) {
            ServerHttpResponse response = exchange.getResponse();
            response.setStatusCode(HttpStatus.TOO_MANY_REQUESTS);
            response.getHeaders().setContentType(MediaType.APPLICATION_JSON);
            
            String body = "{\"code\":429,\"message\":\"请求被限流\"}";
            return response.writeWith(Mono.just(
                response.bufferFactory().wrap(body.getBytes())
            ));
        }
        return Mono.error(ex);
    }
}

4. 验证测试
   • 配置限流规则
   • 发起超过阈值的请求
   • 确认限流生效且不会调用下游服务

避免的误区

• 不要简单地回退 reactor-core 版本，这可能引入安全漏洞
• 不要仅依赖 cancel() 状态，需要显式的错误传播
• 不要在业务代码中捕获 BlockException，应该让它传播到全局异常处理器