feat(reactive-routes,rest): fix CDI request ctx deactivation race

Author: michalvavrik

extensions/reactive-routes/runtime/src/main/java/io/quarkus/vertx/web/runtime/RouteHandler.java

@@ -8,6 +8,7 @@
 import io.quarkus.security.identity.CurrentIdentityAssociation;
 import io.quarkus.security.identity.SecurityIdentity;
 import io.quarkus.vertx.http.runtime.CurrentVertxRequest;
+import io.quarkus.vertx.http.runtime.RoutingUtils;
 import io.quarkus.vertx.http.runtime.security.QuarkusHttpUser;
 import io.quarkus.vertx.web.Route;
 import io.vertx.core.AsyncResult;
@@ -22,7 +23,7 @@
 public abstract class RouteHandler implements Handler<RoutingContext> {
 
     private static final String REQUEST_CONTEXT_STATE = "__cdi_req_ctx";
-
+    private static final String REACTIVE_ROUTES_KEY = "quarkus-reactive-routes";
     private final Event<SecurityIdentity> securityIdentityEvent;
     private final CurrentIdentityAssociation currentIdentityAssociation;
     private final CurrentVertxRequest currentVertxRequest;
@@ -69,6 +70,7 @@ public void handle(RoutingContext context) {
                 ContextState state = context.get(REQUEST_CONTEXT_STATE);
                 // Activate the context, i.e. set the thread locals, state can be null
                 requestContext.activate(state);
+                RoutingUtils.assumeCdiRequestContext(context, REACTIVE_ROUTES_KEY);
                 currentVertxRequest.setCurrent(context);
                 if (currentIdentityAssociation != null) {
                     if (user != null) {
@@ -90,14 +92,18 @@ public void handle(RoutingContext context) {
                     context.addEndHandler(new Handler<AsyncResult<Void>>() {
                         @Override
                         public void handle(AsyncResult<Void> result) {
-                            requestContext.destroy(endState);
+                            if (RoutingUtils.isCdiRequestContextOwner(context, REACTIVE_ROUTES_KEY)) {
+                                requestContext.destroy(endState);
+                            }
                         }
                     });
                 }
                 invoke(context);
             } finally {
-                // Deactivate the context, i.e. cleanup the thread locals
-                requestContext.deactivate();
+                if (RoutingUtils.isCdiRequestContextOwner(context, REACTIVE_ROUTES_KEY)) {
+                    // Deactivate the context, i.e. cleanup the thread locals
+                    requestContext.deactivate();
+                }
             }
         }
     }

extensions/resteasy-reactive/rest-servlet/runtime/src/main/java/io/quarkus/resteasy/reactive/server/servlet/runtime/ServletRequestContext.java

@@ -47,6 +47,7 @@
 import io.quarkus.resteasy.reactive.server.runtime.ResteasyReactiveSecurityContext;
 import io.quarkus.runtime.BlockingOperationControl;
 import io.quarkus.security.identity.SecurityIdentity;
+import io.quarkus.vertx.http.runtime.RoutingUtils;
 import io.quarkus.vertx.http.runtime.security.QuarkusHttpUser;
 import io.undertow.server.HttpServerExchange;
 import io.undertow.server.ResponseCommitListener;
@@ -60,6 +61,7 @@
 public class ServletRequestContext extends ResteasyReactiveRequestContext
         implements ServerHttpRequest, ServerHttpResponse, ResponseCommitListener {
 
+    private static final String QUARKUS_REST_SERVLET_KEY = "quarkus-rest-servlet";
     private static final LazyValue<Event<SecurityIdentity>> SECURITY_IDENTITY_EVENT = new LazyValue<>(
             ServletRequestContext::createEvent);
     final RoutingContext context;
@@ -138,6 +140,11 @@ protected void handleRequestScopeActivation() {
         }
     }
 
+    @Override
+    protected void onPreRequestScopeActivation() {
+        RoutingUtils.assumeCdiRequestContext(context, QUARKUS_REST_SERVLET_KEY);
+    }
+
     static void fireSecurityIdentity(SecurityIdentity identity) {
         SECURITY_IDENTITY_EVENT.get().fire(identity);
     }

extensions/resteasy-reactive/rest/runtime/src/main/java/io/quarkus/resteasy/reactive/server/runtime/QuarkusResteasyReactiveRequestContext.java

@@ -16,13 +16,15 @@
 import io.quarkus.security.identity.CurrentIdentityAssociation;
 import io.quarkus.security.identity.SecurityIdentity;
 import io.quarkus.vertx.core.runtime.context.VertxContextSafetyToggle;
+import io.quarkus.vertx.http.runtime.RoutingUtils;
 import io.quarkus.vertx.http.runtime.security.QuarkusHttpUser;
 import io.smallrye.common.vertx.VertxContext;
 import io.vertx.core.http.impl.HttpServerRequestInternal;
 import io.vertx.ext.web.RoutingContext;
 
 public class QuarkusResteasyReactiveRequestContext extends VertxResteasyReactiveRequestContext {
 
+    private static final String QUARKUS_REST_KEY = "quarkus-rest";
     final CurrentIdentityAssociation association;
     boolean userSetup = false;
 
@@ -51,6 +53,11 @@ protected void handleRequestScopeActivation() {
         }
     }
 
+    @Override
+    protected void onPreRequestScopeActivation() {
+        RoutingUtils.assumeCdiRequestContext(context, QUARKUS_REST_KEY);
+    }
+
     @Override
     protected void requestScopeDeactivated() {
         // we intentionally don't call 'CurrentRequestManager.set(null)'

extensions/vertx-http/runtime/src/main/java/io/quarkus/vertx/http/runtime/RoutingUtils.java

@@ -1,7 +1,10 @@
 package io.quarkus.vertx.http.runtime;
 
+import java.util.Objects;
 import java.util.Set;
 
+import org.jboss.logging.Logger;
+
 import io.vertx.core.http.HttpHeaders;
 import io.vertx.core.http.impl.MimeMapping;
 import io.vertx.core.net.impl.URIDecoder;
@@ -10,10 +13,39 @@
 
 public final class RoutingUtils {
 
+    private static final String CURRENT_CDI_REQUEST_CTX_OWNER = "io.quarkus.vertx.http.runtime#current-cdi-req-ctx-owner";
+    private static final Logger LOG = Logger.getLogger(RoutingUtils.class);
+
     private RoutingUtils() throws IllegalAccessException {
         throw new IllegalAccessException("Avoid direct instantiation");
     }
 
+    /**
+     * Assumes ownership of the currently active CDI request context.
+     * Thus, code invoked (even asynchronously) from previous route handlers shouldn't deactivate it.
+     *
+     * @param ctx RoutingContext
+     * @param newOwner typically a route handler that needs CDI request context active
+     */
+    public static void assumeCdiRequestContext(RoutingContext ctx, String newOwner) {
+        var previousOwner = ctx.data().put(CURRENT_CDI_REQUEST_CTX_OWNER, Objects.requireNonNull(newOwner));
+        if (previousOwner != null && LOG.isDebugEnabled()) {
+            LOG.debugf("CDI request context owner has changed from '%s' to '%s'", previousOwner, newOwner);
+        }
+    }
+
+    /**
+     * Enables route handlers to determine if they can deactivate/destroy CDI request context without impacting
+     * any other extension.
+     *
+     * @param ctx RoutingContext
+     * @param owner typically a route handler that needs CDI request context active
+     * @return true if the CDI request context is owned by the {@code owner}
+     */
+    public static boolean isCdiRequestContextOwner(RoutingContext ctx, String owner) {
+        return owner.equals(ctx.get(CURRENT_CDI_REQUEST_CTX_OWNER));
+    }
+
     /**
      * Get the normalized and decoded path:
      * - normalize based on RFC3986

independent-projects/resteasy-reactive/common/runtime/src/main/java/org/jboss/resteasy/reactive/common/core/AbstractResteasyReactiveContext.java

@@ -246,6 +246,7 @@ public void requireCDIRequestScope() {
         if (requestScopeActivated) {
             return;
         }
+        onPreRequestScopeActivation();
         if (isRequestScopeManagementRequired()) {
             if (requestContext.isRequestContextActive()) {
                 // req. context is already active, just reuse existing one
@@ -274,6 +275,10 @@ public ThreadSetupAction.ThreadState captureCDIRequestScope() {
 
     protected abstract void handleRequestScopeActivation();
 
+    protected void onPreRequestScopeActivation() {
+        // by default do nothing
+    }
+
     /**
      * Restarts handler chain processing on a chain that does not target a specific resource
      * <p>

integration-tests/openapi/src/main/java/io/quarkus/it/openapi/security/FailureStorage.java

@@ -0,0 +1,20 @@
+package io.quarkus.it.openapi.security;
+
+import jakarta.enterprise.context.ApplicationScoped;
+
+import io.quarkus.logging.Log;
+
+@ApplicationScoped
+public class FailureStorage {
+
+    private volatile Throwable throwable = null;
+
+    public Throwable getThrowable() {
+        return throwable;
+    }
+
+    public void setThrowable(Throwable throwable) {
+        Log.info("Setting throwable value to " + throwable);
+        this.throwable = throwable;
+    }
+}

integration-tests/openapi/src/main/java/io/quarkus/it/openapi/security/TestSecurityResource.java

@@ -1,29 +1,91 @@
 package io.quarkus.it.openapi.security;
 
 import jakarta.annotation.security.RolesAllowed;
+import jakarta.enterprise.event.Observes;
+import jakarta.inject.Inject;
+import jakarta.ws.rs.DELETE;
 import jakarta.ws.rs.GET;
 import jakarta.ws.rs.Path;
 import jakarta.ws.rs.core.Context;
+import jakarta.ws.rs.core.HttpHeaders;
 import jakarta.ws.rs.core.SecurityContext;
 
 import io.quarkus.vertx.web.RouteFilter;
+import io.vertx.core.Handler;
+import io.vertx.ext.web.Router;
 import io.vertx.ext.web.RoutingContext;
 
 @Path("/security")
 public class TestSecurityResource {
 
+    public static final String TEST_HEADER_NAME = "test-security-header";
+    public static final String TEST_HEADER_VALUE = "hush-hush";
+    public static int REQUEST_TIMEOUT = 3;
+
+    @Inject
+    FailureStorage failureStorage;
+
+    @Context
+    HttpHeaders httpHeaders;
+
     @RolesAllowed("admin")
     @GET
     @Path("reactive-routes")
     public String reactiveRoutes(@Context SecurityContext securityContext) {
         return securityContext.getUserPrincipal().getName();
     }
 
+    @RolesAllowed("admin")
+    @GET
+    @Path("reactive-routes-with-delayed-response")
+    public String reactiveRoutesWithDelayedResponse(@Context SecurityContext securityContext) throws InterruptedException {
+        Thread.sleep(REQUEST_TIMEOUT);
+        try {
+            // ATM this code is not invoked every single time, it is timing issue
+            // but on my laptop it is invoked 7 times out of 10, therefore, if re-run multiple times,
+            // it sufficiently reproduces our original issue with accessing CDI request context
+            var headerValue = httpHeaders.getHeaderString(TEST_HEADER_NAME);
+
+            // just to do something - check the expected value
+            if (!TEST_HEADER_VALUE.equals(headerValue)) {
+                throw new IllegalStateException(
+                        "Invalid header value, got '%s',but expected '%s' ".formatted(headerValue, TEST_HEADER_VALUE));
+            }
+        } catch (Throwable t) {
+            failureStorage.setThrowable(t);
+        }
+        return securityContext.getUserPrincipal().getName();
+    }
+
+    @Path("throwable")
+    @GET
+    public String getThrowable() {
+        return String.valueOf(failureStorage.getThrowable());
+    }
+
+    @Path("empty-failure-storage")
+    @DELETE
+    public void emptyFailureStorage() {
+        failureStorage.setThrowable(null);
+    }
+
     @RouteFilter(401)
     public void doNothing(RoutingContext routingContext) {
         // here so that the Reactive Routes extension activates CDI request context
         routingContext.response().putHeader("reactive-routes-filter", "true");
         routingContext.next();
     }
 
+    void addFailureHandler(@Observes Router router) {
+        // this is necessary, because before the fix, the ContextNotActiveException was sometimes
+        // thrown in the CDI interceptors and handled by the QuarkusErrorHandler
+        router.route().order(Integer.MAX_VALUE - 100).failureHandler(new Handler<RoutingContext>() {
+            @Override
+            public void handle(RoutingContext routingContext) {
+                failureStorage.setThrowable(routingContext.failure());
+                routingContext.end();
+            }
+        });
+    }
+
 }

integration-tests/openapi/src/test/java/io/quarkus/it/openapi/security/TestSecurityReactiveRoutesTest.java

@@ -1,13 +1,21 @@
 package io.quarkus.it.openapi.security;
 
+import static io.quarkus.it.openapi.security.TestSecurityResource.TEST_HEADER_NAME;
+import static io.quarkus.it.openapi.security.TestSecurityResource.TEST_HEADER_VALUE;
+import static org.apache.http.params.CoreConnectionPNames.SO_TIMEOUT;
 import static org.hamcrest.Matchers.is;
 
+import java.net.SocketTimeoutException;
+import java.time.Duration;
+
+import org.assertj.core.api.Assertions;
 import org.junit.jupiter.api.Test;
 
 import io.quarkus.test.common.http.TestHTTPEndpoint;
 import io.quarkus.test.junit.QuarkusTest;
 import io.quarkus.test.security.TestSecurity;
 import io.restassured.RestAssured;
+import io.restassured.config.HttpClientConfig;
 
 @QuarkusTest
 @TestHTTPEndpoint(TestSecurityResource.class)
@@ -23,4 +31,47 @@ public void testSecurityWithReactiveRoutesAndQuarkusRest() {
                 .body(is("Martin"));
     }
 
+    /**
+     * This verifies that CDI request context activated by Reactive Routes is not deactivated/destroyed while
+     * Quarkus REST needs it. Before the fix, there was a racy behavior. Sometimes during the CDI interceptors processing,
+     * sometimes after the socket timeout when resource method was executed, CDI request context was not active.
+     * Depending on the speed of a test executor, you may need to execute this test couple of times in order to reproduce
+     * the original issue.
+     */
+    @TestSecurity(user = "Martin", roles = "admin")
+    @Test
+    public void testCdiRequestActiveAfterTimeout() throws InterruptedException {
+        RestAssured.delete("empty-failure-storage").then().statusCode(204);
+
+        int valueLesserThanDelay = TestSecurityResource.REQUEST_TIMEOUT - 2;
+        // using deprecated constant due to https://github.com/rest-assured/rest-assured/issues/497
+        var config = RestAssured.config()
+                .httpClient(HttpClientConfig.httpClientConfig().setParam(SO_TIMEOUT, valueLesserThanDelay));
+        try {
+            RestAssured.given()
+                    .config(config)
+                    .header(TEST_HEADER_NAME, TEST_HEADER_VALUE)
+                    .get("reactive-routes-with-delayed-response")
+                    .then()
+                    .statusCode(200)
+                    .header("reactive-routes-filter", is("true"))
+                    .body(is("Martin"));
+            Assertions.fail("HTTP request didn't result in a socket timeout exception");
+        } catch (Exception socketTimeoutException) {
+            // yes, this checked exception is thrown even though no method signature declares it
+            if (!(socketTimeoutException instanceof SocketTimeoutException)) {
+                // socket timeout exception is required to verify what happens with the CDI request context after the timeout
+                Assertions.fail("Expected a SocketTimeoutException but got " + socketTimeoutException);
+            }
+        }
+        int timeoutRemainder = TestSecurityResource.REQUEST_TIMEOUT - valueLesserThanDelay + 1;
+        Thread.sleep(Duration.ofSeconds(timeoutRemainder).toMillis());
+        RestAssured.given()
+                .get("throwable")
+                .then()
+                .statusCode(200)
+                .header("reactive-routes-filter", is("true"))
+                .body(is("null"));
+    }
+
 }