fix hanging requests and filters read with the wrong CRS under concurrent requests

xtraplatform-cql/src/main/java/de/ii/xtraplatform/cql/app/CqlImpl.java

@@ -8,6 +8,8 @@
 package de.ii.xtraplatform.cql.app;
 
 import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.databind.BeanProperty;
+import com.fasterxml.jackson.databind.DeserializationContext;
 import com.fasterxml.jackson.databind.InjectableValues;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.SerializationFeature;
@@ -50,6 +52,11 @@ public class CqlImpl implements Cql {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(CqlImpl.class);
 
+  // The CRS of the coordinates in a filter is injected into the deserializers. The mapper is
+  // shared by all requests, so the CRS is bound to the thread that reads the filter.
+  private static final ThreadLocal<Optional<EpsgCrs>> FILTER_CRS =
+      ThreadLocal.withInitial(Optional::empty);
+
   private final CqlTextParser cqlTextParser;
   private final ObjectMapper cqlJsonMapper;
 
@@ -67,7 +74,34 @@ public CqlImpl() {
             .setSerializationInclusion(JsonInclude.Include.NON_ABSENT)
             .enable(SerializationFeature.INDENT_OUTPUT)
             .disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS)
-            .registerModule(module);
+            .registerModule(module)
+            .setInjectableValues(new FilterCrsInjectableValues());
+  }
+
+  private static final class FilterCrsInjectableValues extends InjectableValues {
+
+    @Override
+    public Object findInjectableValue(
+        DeserializationContext ctxt,
+        Object valueId,
+        BeanProperty forProperty,
+        Object beanInstance,
+        Boolean optional,
+        Boolean useInput) {
+      return findInjectableValue(valueId, ctxt, forProperty, beanInstance);
+    }
+
+    @Override
+    public Object findInjectableValue(
+        Object valueId,
+        DeserializationContext ctxt,
+        BeanProperty forProperty,
+        Object beanInstance) {
+      if ("filterCrs".equals(valueId)) {
+        return FILTER_CRS.get();
+      }
+      throw new IllegalArgumentException(String.format("No injectable value '%s'", valueId));
+    }
   }
 
   @Override
@@ -87,8 +121,7 @@ public Cql2Expression read(String cql, Format format, EpsgCrs crs, boolean allow
     }
 
     if (format == Format.JSON) {
-      cqlJsonMapper.setInjectableValues(
-          new InjectableValues.Std().addValue("filterCrs", Optional.ofNullable(crs)));
+      FILTER_CRS.set(Optional.ofNullable(crs));
       try {
         Cql2Expression expression = cqlJsonMapper.readValue(cql, Cql2Expression.class);
 
@@ -103,6 +136,8 @@ public Cql2Expression read(String cql, Format format, EpsgCrs crs, boolean allow
         return expression;
       } catch (IOException e) {
         throw new CqlParseException(e.getMessage(), e);
+      } finally {
+        FILTER_CRS.remove();
       }
     }
 

xtraplatform-cql/src/test/groovy/de/ii/xtraplatform/cql/app/CqlJsonSpec.groovy

@@ -9,10 +9,20 @@ package de.ii.xtraplatform.cql.app
 
 import de.ii.xtraplatform.cql.domain.Cql
 import de.ii.xtraplatform.cql.domain.Cql2Expression
+import de.ii.xtraplatform.cql.domain.BinarySpatialOperation
+import de.ii.xtraplatform.cql.domain.GeometryNode
+import de.ii.xtraplatform.cql.domain.SpatialLiteral
+import de.ii.xtraplatform.crs.domain.EpsgCrs
+import de.ii.xtraplatform.crs.domain.OgcCrs
 import org.skyscreamer.jsonassert.JSONAssert
 import spock.lang.Shared
 import spock.lang.Specification
 
+import java.util.concurrent.Callable
+import java.util.concurrent.ExecutorService
+import java.util.concurrent.Executors
+import java.util.concurrent.Future
+
 class CqlJsonSpec extends Specification {
 
     @Shared
@@ -1370,4 +1380,27 @@ class CqlJsonSpec extends Specification {
         then:
         JSONAssert.assertEquals(cqlJson, actual2, true)
     }
+
+    def 'concurrent reads keep the CRS of each filter'() {
+        given:
+        String filter = '{"op":"s_intersects","args":[{"property":"geometry"},{"type":"Point","coordinates":[455000,5543000]}]}'
+        List<EpsgCrs> crsList = (0..<200).collect { it % 2 == 0 ? EpsgCrs.of(25832) : OgcCrs.CRS84 }
+        ExecutorService executor = Executors.newFixedThreadPool(8)
+
+        when:
+        List<Future<EpsgCrs>> results = crsList.collect { crs ->
+            executor.submit({
+                Cql2Expression expression = cql.read(filter, Cql.Format.JSON, crs)
+                SpatialLiteral literal = (SpatialLiteral) ((BinarySpatialOperation) expression).getArgs().get(1)
+                ((GeometryNode) literal.getValue()).getGeometry().getCrs().orElse(null)
+            } as Callable<EpsgCrs>)
+        }
+        List<EpsgCrs> actual = results.collect { it.get() }
+
+        then:
+        actual == crsList
+
+        cleanup:
+        executor.shutdownNow()
+    }
 }

xtraplatform-features/src/main/java/de/ii/xtraplatform/features/domain/FeatureStreamImpl.java

@@ -106,6 +106,27 @@ public FeatureStreamImpl(
     this.idsArePerType = !data.getGloballyUniqueFeatureIds();
   }
 
+  // The collection metadata is published when the stream starts. A stream that ends before, e.g.
+  // because the database rejects the first query, completes it with the stream error; a caller
+  // waiting for the metadata would block forever otherwise.
+  static <W extends ResultBase> CompletionStage<W> completeMetadataOnEnd(
+      CompletionStage<W> result, CompletableFuture<CollectionMetadata> onCollectionMetadata) {
+    return result.whenComplete(
+        (r, throwable) -> {
+          if (onCollectionMetadata.isDone()) {
+            return;
+          }
+          Throwable error =
+              Objects.nonNull(throwable)
+                  ? throwable
+                  : Objects.nonNull(r) ? r.getError().orElse(null) : null;
+          onCollectionMetadata.completeExceptionally(
+              Objects.requireNonNullElseGet(
+                  error,
+                  () -> new IllegalStateException("The feature stream ended before it started.")));
+        });
+  }
+
   // For types without properties that are represented as links (an explicit `link` in the
   // schema or a role that declares a link relation) the PropertyLinks transformer would be a
   // per-token no-op and is not wired at all.
@@ -235,7 +256,8 @@ public CompletionStage<Result> runWith(
                   });
         };
 
-    return runner.runQuery(stream, query, mergedTransformations, !doTransform);
+    return completeMetadataOnEnd(
+        runner.runQuery(stream, query, mergedTransformations, !doTransform), onCollectionMetadata);
   }
 
   @Override
@@ -336,7 +358,8 @@ public <X> CompletionStage<ResultReduced<X>> runWith(
                   });
         };
 
-    return runner.runQuery(stream, query, mergedTransformations, !doTransform);
+    return completeMetadataOnEnd(
+        runner.runQuery(stream, query, mergedTransformations, !doTransform), onCollectionMetadata);
   }
 
   private FeatureTokenSource getFeatureTokenSourceTransformed(

xtraplatform-features/src/test/groovy/de/ii/xtraplatform/features/domain/FeatureStreamMetadataSpec.groovy

@@ -0,0 +1,78 @@
+/*
+ * Copyright 2026 interactive instruments GmbH
+ *
+ * This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, You can obtain one at http://mozilla.org/MPL/2.0/.
+ */
+package de.ii.xtraplatform.features.domain
+
+import spock.lang.Specification
+
+import java.util.concurrent.CompletableFuture
+import java.util.concurrent.CompletionException
+import java.util.OptionalLong
+
+class FeatureStreamMetadataSpec extends Specification {
+
+    def 'a stream that ends with an error before it started completes the metadata with the error'() {
+        given:
+        def error = new IllegalStateException("rejected by the database")
+        def result = CompletableFuture.completedFuture(
+                ImmutableResult.builder().isEmpty(true).hasFeatures(false).error(error).build())
+        def metadata = new CompletableFuture<CollectionMetadata>()
+
+        when:
+        FeatureStreamImpl.completeMetadataOnEnd(result, metadata).toCompletableFuture().join()
+        metadata.join()
+
+        then:
+        def e = thrown(CompletionException)
+        e.cause.is(error)
+    }
+
+    def 'a stream that fails before it started completes the metadata with the failure'() {
+        given:
+        def error = new IllegalArgumentException("no connector")
+        def result = CompletableFuture.<FeatureStream.Result> failedFuture(error)
+        def metadata = new CompletableFuture<CollectionMetadata>()
+
+        when:
+        FeatureStreamImpl.completeMetadataOnEnd(result, metadata)
+        metadata.join()
+
+        then:
+        def e = thrown(CompletionException)
+        e.cause.is(error)
+    }
+
+    def 'a stream that ends without an error before it started completes the metadata exceptionally'() {
+        given:
+        def result = CompletableFuture.completedFuture(
+                ImmutableResult.builder().isEmpty(true).hasFeatures(false).build())
+        def metadata = new CompletableFuture<CollectionMetadata>()
+
+        when:
+        FeatureStreamImpl.completeMetadataOnEnd(result, metadata).toCompletableFuture().join()
+        metadata.join()
+
+        then:
+        def e = thrown(CompletionException)
+        e.cause instanceof IllegalStateException
+    }
+
+    def 'metadata published when the stream started is kept'() {
+        given:
+        def published = ImmutableCollectionMetadata.of(OptionalLong.of(0L), OptionalLong.of(0L))
+        def metadata = CompletableFuture.<CollectionMetadata> completedFuture(published)
+        def error = new IllegalStateException("failed while streaming")
+        def result = CompletableFuture.completedFuture(
+                ImmutableResult.builder().isEmpty(true).hasFeatures(false).error(error).build())
+
+        when:
+        FeatureStreamImpl.completeMetadataOnEnd(result, metadata).toCompletableFuture().join()
+
+        then:
+        metadata.join().is(published)
+    }
+}