CELDEV-1343 EndpointConcurrencyLimiter

celements-spring-mvc/pom.xml

@@ -38,6 +38,14 @@
       <groupId>org.slf4j</groupId>
       <artifactId>slf4j-api</artifactId>
     </dependency>
+    <dependency>
+      <groupId>com.google.guava</groupId>
+      <artifactId>guava</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>one.util</groupId>
+      <artifactId>streamex</artifactId>
+    </dependency>
 
     <dependency>
       <groupId>com.celements</groupId>

celements-spring-mvc/src/main/java/com/celements/spring/mvc/HelloWorldController.java

@@ -1,40 +1,18 @@
 package com.celements.spring.mvc;
 
-import javax.inject.Inject;
-
-import org.springframework.beans.factory.BeanFactory;
 import org.springframework.security.access.prepost.PreAuthorize;
 import org.springframework.web.bind.annotation.GetMapping;
 import org.springframework.web.bind.annotation.RestController;
-import org.xwiki.context.Execution;
 
-import com.celements.execution.XWikiExecutionProp;
 import com.celements.spring.security.AuthenticatedBaseController;
-import com.xpn.xwiki.XWikiContext;
 
 @RestController
 public class HelloWorldController extends AuthenticatedBaseController {
 
-  private final BeanFactory beanFactory;
-
-  @Inject
-  public HelloWorldController(BeanFactory beanFactory) {
-    this.beanFactory = beanFactory;
-  }
-
-  @GetMapping("/helloworld")
+  @GetMapping
   @PreAuthorize("permitAll()")
   public String helloWorld() {
     return "Hello World!";
   }
 
-  @GetMapping("/hellocontext")
-  public XWikiContext helloContext() {
-    return beanFactory
-        .getBean(Execution.class)
-        .getContext()
-        .get(XWikiExecutionProp.XWIKI_CONTEXT)
-        .orElseThrow();
-  }
-
 }

celements-spring-mvc/src/main/java/com/celements/spring/mvc/limiter/EndpointConcurrencyConfig.java

@@ -0,0 +1,11 @@
+package com.celements.spring.mvc.limiter;
+
+import static com.google.common.base.Preconditions.*;
+
+public record EndpointConcurrencyConfig(int maxConcurrent, int waitMillis) {
+
+  public EndpointConcurrencyConfig {
+    checkArgument(maxConcurrent > 0, "Endpoint concurrency limit must be positive");
+    checkArgument(waitMillis >= 0, "Endpoint concurrency wait time must be non-negative");
+  }
+}

celements-spring-mvc/src/main/java/com/celements/spring/mvc/limiter/EndpointConcurrencyLimit.java

@@ -0,0 +1,16 @@
+package com.celements.spring.mvc.limiter;
+
+import static java.lang.annotation.ElementType.*;
+import static java.lang.annotation.RetentionPolicy.*;
+
+import java.lang.annotation.Documented;
+import java.lang.annotation.Retention;
+import java.lang.annotation.Target;
+
+@Documented
+@Retention(RUNTIME)
+@Target({ TYPE, METHOD })
+public @interface EndpointConcurrencyLimit {
+
+  String value();
+}

celements-spring-mvc/src/main/java/com/celements/spring/mvc/limiter/EndpointConcurrencyLimiter.java

@@ -0,0 +1,84 @@
+package com.celements.spring.mvc.limiter;
+
+import static com.google.common.base.Preconditions.*;
+import static java.util.concurrent.TimeUnit.*;
+import static org.springframework.core.annotation.AnnotatedElementUtils.*;
+import static org.springframework.http.HttpStatus.*;
+
+import java.io.IOException;
+import java.lang.reflect.AnnotatedElement;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.Semaphore;
+
+import javax.inject.Inject;
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+
+import org.springframework.stereotype.Component;
+import org.springframework.web.method.HandlerMethod;
+
+import com.celements.spring.mvc.CelMvcInterceptor;
+
+import one.util.streamex.EntryStream;
+
+@Component
+public class EndpointConcurrencyLimiter implements CelMvcInterceptor {
+
+  private final Map<String, EndpointConcurrencyConfig> configs;
+  private final Map<String, Semaphore> limiters;
+
+  @Inject
+  public EndpointConcurrencyLimiter(Optional<Map<String, EndpointConcurrencyConfig>> configs) {
+    this(configs.orElseGet(Map::of));
+  }
+
+  EndpointConcurrencyLimiter(Map<String, EndpointConcurrencyConfig> configs) {
+    this.configs = Map.copyOf(configs);
+    this.limiters = EntryStream.of(configs)
+        .mapValues(config -> new Semaphore(config.maxConcurrent()))
+        .toMap();
+  }
+
+  @Override
+  public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler)
+      throws IOException {
+    if (findLimiterName(handler).map(this::tryAcquire).orElse(true)) {
+      return true;
+    }
+    response.sendError(SERVICE_UNAVAILABLE.value(), "Endpoint concurrency limit exhausted");
+    return false;
+  }
+
+  private boolean tryAcquire(String limiterName) {
+    try {
+      var config = configs.get(limiterName);
+      checkState(config != null, "Missing EndpointConcurrencyConfig [%s]", limiterName);
+      return limiters.get(limiterName).tryAcquire(config.waitMillis(), MILLISECONDS);
+    } catch (InterruptedException exc) {
+      Thread.currentThread().interrupt();
+      return false;
+    }
+  }
+
+  @Override
+  public void afterCompletion(HttpServletRequest request, HttpServletResponse response,
+      Object handler, Exception exc) {
+    findLimiterName(handler)
+        .map(limiters::get)
+        .ifPresent(Semaphore::release);
+  }
+
+  private Optional<String> findLimiterName(Object handler) {
+    if (!(handler instanceof HandlerMethod methodHandler)) {
+      return Optional.empty();
+    }
+    return findAnnotation(methodHandler.getMethod())
+        .or(() -> findAnnotation(methodHandler.getBeanType()))
+        .map(EndpointConcurrencyLimit::value);
+  }
+
+  private Optional<EndpointConcurrencyLimit> findAnnotation(AnnotatedElement element) {
+    return Optional.ofNullable(findMergedAnnotation(element, EndpointConcurrencyLimit.class));
+  }
+}

celements-spring-mvc/src/test/java/com/celements/spring/mvc/limiter/EndpointConcurrencyLimiterTest.java

@@ -0,0 +1,163 @@
+package com.celements.spring.mvc.limiter;
+
+import static org.junit.Assert.*;
+import static org.springframework.http.HttpStatus.*;
+import static java.util.concurrent.Executors.*;
+import static java.util.concurrent.TimeUnit.*;
+
+import java.util.Map;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.springframework.beans.factory.config.ConfigurableBeanFactory;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Primary;
+import org.springframework.context.annotation.Scope;
+import org.springframework.mock.web.MockHttpServletRequest;
+import org.springframework.mock.web.MockHttpServletResponse;
+import org.springframework.web.method.HandlerMethod;
+
+import com.celements.common.test.AbstractComponentTest;
+import com.celements.spring.mvc.limiter.EndpointConcurrencyConfig;
+import com.celements.spring.mvc.limiter.EndpointConcurrencyLimit;
+import com.celements.spring.mvc.limiter.EndpointConcurrencyLimiter;
+
+public class EndpointConcurrencyLimiterTest extends AbstractComponentTest {
+
+  private EndpointConcurrencyLimiter limiter;
+
+  @Before
+  public void prepareTest() {
+    limiter = getBeanFactory().getBean(EndpointConcurrencyLimiter.class);
+  }
+
+  @Test
+  public void test_unannotatedHandlersAreUnlimited() throws Exception {
+    assertTrue(limiter.preHandle(new MockHttpServletRequest(), new MockHttpServletResponse(),
+        handlerMethod("unlimitedEndpoint")));
+    assertTrue(limiter.preHandle(
+        new MockHttpServletRequest(), new MockHttpServletResponse(), new Object()));
+  }
+
+  @Test
+  public void test_sameLimitSharesCapacityAcrossEndpoints() throws Exception {
+    var request = new MockHttpServletRequest();
+    var handler = handlerMethod("limitedEndpoint");
+    var otherHandler = handlerMethod("otherSearchEndpoint");
+    assertTrue(limiter.preHandle(request, new MockHttpServletResponse(), handler));
+
+    var response = new MockHttpServletResponse();
+    assertFalse(limiter.preHandle(new MockHttpServletRequest(), response, otherHandler));
+    assertEquals(SERVICE_UNAVAILABLE.value(), response.getStatus());
+
+    limiter.afterCompletion(request, null, handler, null);
+    assertTrue(limiter.preHandle(new MockHttpServletRequest(), new MockHttpServletResponse(),
+        otherHandler));
+  }
+
+  @Test
+  public void test_waitsForCapacity() throws Exception {
+    var waitingLimiter = new EndpointConcurrencyLimiter(Map.of(
+        "search", new EndpointConcurrencyConfig(1, 1000)));
+    var request = new MockHttpServletRequest();
+    var handler = handlerMethod("limitedEndpoint");
+    assertTrue(waitingLimiter.preHandle(request, new MockHttpServletResponse(), handler));
+
+    var executor = newSingleThreadScheduledExecutor();
+    try {
+      executor.schedule(
+          () -> waitingLimiter.afterCompletion(request, null, handler, null), 50, MILLISECONDS);
+      assertTrue(waitingLimiter.preHandle(
+          new MockHttpServletRequest(), new MockHttpServletResponse(), handler));
+    } finally {
+      executor.shutdownNow();
+    }
+  }
+
+  @Test
+  public void test_differentLimitsHaveIndependentCapacity() throws Exception {
+    assertTrue(limiter.preHandle(new MockHttpServletRequest(), new MockHttpServletResponse(),
+        handlerMethod("limitedEndpoint")));
+    assertTrue(limiter.preHandle(new MockHttpServletRequest(), new MockHttpServletResponse(),
+        handlerMethod("otherLimitedEndpoint")));
+  }
+
+  @Test
+  public void test_missingConfigFailsClosed() throws Exception {
+    assertThrows(IllegalStateException.class, () -> limiter.preHandle(
+        new MockHttpServletRequest(), new MockHttpServletResponse(),
+        handlerMethod("missingLimitEndpoint")));
+  }
+
+  @Test
+  public void test_classAnnotation() throws Exception {
+    var request = new MockHttpServletRequest();
+    var handler = new HandlerMethod(
+        new LimitedController(), LimitedController.class.getMethod("limitedEndpoint"));
+
+    assertTrue(limiter.preHandle(request, new MockHttpServletResponse(), handler));
+    assertFalse(limiter.preHandle(
+        new MockHttpServletRequest(), new MockHttpServletResponse(), handler));
+  }
+
+  @Test
+  public void test_configRejectsNonPositiveConcurrency() {
+    assertThrows(IllegalArgumentException.class, () -> new EndpointConcurrencyConfig(0, 0));
+    assertThrows(IllegalArgumentException.class, () -> new EndpointConcurrencyConfig(-1, 0));
+    assertThrows(IllegalArgumentException.class, () -> new EndpointConcurrencyConfig(1, -1));
+  }
+
+  private HandlerMethod handlerMethod(String name) {
+    try {
+      return new HandlerMethod(new TestController(), TestController.class.getMethod(name));
+    } catch (NoSuchMethodException exc) {
+      throw new IllegalArgumentException(exc);
+    }
+  }
+
+  private static class TestController {
+
+    public void unlimitedEndpoint() {}
+
+    @EndpointConcurrencyLimit("search")
+    public void limitedEndpoint() {}
+
+    @EndpointConcurrencyLimit("search")
+    public void otherSearchEndpoint() {}
+
+    @EndpointConcurrencyLimit("other")
+    public void otherLimitedEndpoint() {}
+
+    @EndpointConcurrencyLimit("missing")
+    public void missingLimitEndpoint() {}
+  }
+
+  @EndpointConcurrencyLimit("search")
+  private static class LimitedController {
+
+    public void limitedEndpoint() {}
+  }
+
+  @Configuration
+  static class TestConfig {
+
+    @Bean("search")
+    EndpointConcurrencyConfig searchLimiterConfig() {
+      return new EndpointConcurrencyConfig(1, 0);
+    }
+
+    @Bean("other")
+    EndpointConcurrencyConfig otherLimiterConfig() {
+      return new EndpointConcurrencyConfig(1, 0);
+    }
+
+    @Bean
+    @Primary
+    @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
+    EndpointConcurrencyLimiter endpointConcurrencyLimiter(
+        Map<String, EndpointConcurrencyConfig> configs) {
+      return new EndpointConcurrencyLimiter(configs);
+    }
+  }
+}