Recursive subscribe/loop during service shutdown

[EricY019]

Description

During a Spring context close, a pipeline that uses Mono.when(...) can intermittently enter a recursive subscribe/loop. The logs repeatedly show MonoWhen$WhenInner.currentContext alternating with Operators.reportThrowInSubscribe and Mono.subscribe, suggesting a subscription cycle during termination.

Expected Behavior

Shutdown should complete outstanding signals and release resources in order.
No recursive calls to currentContext/subscribe or subscription loops should occur.

Actual Behavior

Repeated stack traces during shutdown indicate recursive/looping subscriptions.

Steps to Reproduce

@Test
void reproCase() {

}

Possible Solution

Your Environment

• Reactor version(s) used: 3.7.8
• Other relevant libraries versions (eg. netty, ...):
• JVM version (java -version):
  openjdk version "21.0.7" 2025-04-15 LTS
  OpenJDK Runtime Environment Zulu21.42+19-CA (build 21.0.7+6-LTS)
  OpenJDK 64-Bit Server VM Zulu21.42+19-CA (build 21.0.7+6-LTS, mixed mode, sharing)
• OS and version (eg uname -a): x86_64 GNU/Linux

[EricY019]

Steps to Reproduce

Minimal controller snippet

• Rate-limit gate: Mono<Boolean>
• Parameter validation
• getMatcherIdByInstId resolution
• onDepthQuery bridged via Mono.create(...)
• Global request timeout around 1000ms
• Error mapping via .onErrorResume(...)

@RestController
public class ReproController {

    private static final int MAX_DEPTH_LEVEL = 5000;
    private static final int DEFAULT_SIZE = 1;

    @Value("${timeout-ms:1000}")
    private int depthQueryTimeoutMs;

    private final RateLimitService rateLimitService;   // CORE: rate limit (Mono<Boolean>)
    private final LauncherLike launcher;               // CORE: getMatcherIdByInstId + onDepthQuery callback entry

    public ReproController(RateLimitService rateLimitService, LauncherLike launcher) {
        this.rateLimitService = rateLimitService;
        this.launcher = launcher;
    }

    @GetMapping(value = "/api/v5/market/books-full", produces = MediaType.APPLICATION_JSON_VALUE)
    public Mono<String> booksFull(
            final ServerHttpRequest request,
            final @RequestParam("instId") String instId,
            final @RequestParam(value = "sz", required = false) String size
    ) {
        final String clientIp = realIp(request);
        final Mono<Boolean> rateLimit = rateLimitService.rateLimit(clientIp);

        return rateLimit
                // rate-limit check
                .map(allowed -> {
                    if (!allowed) throw new RateLimitedException(clientIp);
                    return true;
                })
                // parameter validation (sz in [1, 5000])
                .map(__ -> validateSizeOrThrow(size))
                // matcherId resolution
                .flatMap(__ -> {
                    int matcherId = launcher.getMatcherIdByInstId(instId);
                    if (matcherId == -1) throw new InstIdNotFoundException(instId);
                    return Mono.just(matcherId);
                })
                // depth query (callback → Mono bridge)
                .flatMap(matcherId -> {
                    int parsedSize = size == null ? DEFAULT_SIZE : Integer.parseInt(size);
                    return getDepth("BOOKS_5000_QUERY_NAME", matcherId, parsedSize, "books-full");
                })
                // global timeout (~1000ms)
                .timeout(Duration.ofMillis(depthQueryTimeoutMs))
                // error mapping
                .onErrorResume(TimeoutException.class, e ->
                        Mono.just("{\"code\":\"50026\",\"msg\":\"System error. Try again later.\"}"))
                .onErrorResume(InstIdNotFoundException.class, e ->
                        Mono.just("{\"code\":\"51001\",\"msg\":\"Instrument ID does not exist.\"}"))
                .onErrorResume(SzInvalidException.class, e ->
                        Mono.just("{\"code\":\"50015\",\"msg\":\"Parameter sz error\"}"))
                .onErrorResume(RateLimitedException.class, e ->
                        Mono.just("{\"code\":\"50011\",\"msg\":\"Rate limit reached. Please refer to API documentation and throttle requests accordingly.\"}"))
                .onErrorResume(e ->
                        Mono.just("{\"code\":\"50026\",\"msg\":\"System error. Try again later.\"}"));
    }

    private boolean validateSizeOrThrow(String size) {
        if (size == null) return true;
        try {
            int v = Integer.parseInt(size);
            if (1 <= v && v <= MAX_DEPTH_LEVEL) return true;
        } catch (NumberFormatException ignore) { }
        throw new SzInvalidException(size);
    }

    private Mono<String> getDepth(String queryName, int matcherId, int size, String source) {
        return Mono.<String>create(sink -> {
                    QueryCallback cb = QueryCallback.fromSink(sink);
                    launcher.onDepthQuery(queryName, matcherId, size, cb, true);
                })
                .map(data -> "{\"result\":\"" + data + "\"}");
    }

    private String realIp(ServerHttpRequest req) {
        String v = req.getHeaders().getFirst("X-Real-IP");
        return v != null ? v : (req.getRemoteAddress() != null ? req.getRemoteAddress().getAddress().getHostAddress() : "0.0.0.0");
    }

    interface RateLimitService { Mono<Boolean> rateLimit(String key); }
    interface LauncherLike {
        int getMatcherIdByInstId(String instId);
        void onDepthQuery(String queryName, int matcherId, int size, QueryCallback cb, boolean trace);
    }
    static class RateLimitedException extends RuntimeException { RateLimitedException(String ip){super(ip);} }
    static class SzInvalidException extends RuntimeException { SzInvalidException(String sz){super(sz);} }
    static class InstIdNotFoundException extends RuntimeException { InstIdNotFoundException(String id){super(id);} }

    interface QueryCallback {
        boolean isCancelled();
        void onSuccess(String value);
        void onError(Throwable e);
        static QueryCallback fromSink(reactor.core.publisher.MonoSink<String> sink) {
            java.util.concurrent.atomic.AtomicBoolean cancelled = new java.util.concurrent.atomic.AtomicBoolean(false);
            sink.onCancel(() -> cancelled.set(true));
            return new QueryCallback() {
                public boolean isCancelled() { return cancelled.get(); }
                public void onSuccess(String v) { if (!cancelled.get()) sink.success(v); }
                public void onError(Throwable e) { if (!cancelled.get()) sink.error(e); }
            };
        }
    }
}

Steps

1. Start a Spring Boot WebFlux app using Reactor Core 3.7.8 and the controller above (timeout ≈1000ms).
2. Begin sustained load (2k–8k TPS) to keep thousands of requests in flight for at least 60s.
3. While load is actively running (after ~5–10s stabilization), send SIGTERM to the Java process to trigger graceful shutdown:
   • kill -15 <PID>
4. Observe logs during shutdown:
   • Intermittently, repeated stack frames appear, alternating between:
     • reactor.core.publisher.MonoWhen$WhenInner.currentContext
     • reactor.core.publisher.Operators.reportThrowInSubscribe
     • reactor.core.publisher.Mono.subscribe
   • Shutdown may be delayed; transient CPU spikes may occur.