From 534409e65caec339db200f40804d35caf1489e91 Mon Sep 17 00:00:00 2001 From: Clemens Portele Date: Thu, 15 Oct 2026 13:43:33 +0200 Subject: [PATCH 1/2] fix filter geometries in the wrong CRS under concurrent requests The CRS of the coordinates in a CQL2-JSON filter was set on the shared object mapper before the filter was read. With concurrent requests, a filter could be read with the CRS of another request, e.g. a stored query with coordinates in EPSG:25832 was read as CRS84. The coordinates were then rejected as invalid, which failed the request with an internal error, also after a response had already been started. The CRS is now bound to the thread that reads the filter. --- .../de/ii/xtraplatform/cql/app/CqlImpl.java | 41 +++++++++++++++++-- .../xtraplatform/cql/app/CqlJsonSpec.groovy | 33 +++++++++++++++ 2 files changed, 71 insertions(+), 3 deletions(-) diff --git a/xtraplatform-cql/src/main/java/de/ii/xtraplatform/cql/app/CqlImpl.java b/xtraplatform-cql/src/main/java/de/ii/xtraplatform/cql/app/CqlImpl.java index e5de1cdf2..b67aa056e 100644 --- a/xtraplatform-cql/src/main/java/de/ii/xtraplatform/cql/app/CqlImpl.java +++ b/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> 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(); } } diff --git a/xtraplatform-cql/src/test/groovy/de/ii/xtraplatform/cql/app/CqlJsonSpec.groovy b/xtraplatform-cql/src/test/groovy/de/ii/xtraplatform/cql/app/CqlJsonSpec.groovy index 480e6b1c3..420b91c8d 100644 --- a/xtraplatform-cql/src/test/groovy/de/ii/xtraplatform/cql/app/CqlJsonSpec.groovy +++ b/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 crsList = (0..<200).collect { it % 2 == 0 ? EpsgCrs.of(25832) : OgcCrs.CRS84 } + ExecutorService executor = Executors.newFixedThreadPool(8) + + when: + List> 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) + } + List actual = results.collect { it.get() } + + then: + actual == crsList + + cleanup: + executor.shutdownNow() + } } From 0ab28376eea05ba6d4ea0e1b36e31e5e05912fc8 Mon Sep 17 00:00:00 2001 From: Clemens Portele Date: Thu, 15 Oct 2026 19:11:01 +0200 Subject: [PATCH 2/2] fix requests hanging when the database rejects the first query of a feature stream The collection metadata of a feature stream is published when the stream starts. When the first query fails (e.g. a filter that the database rejects), the stream ended with the error without ever starting, and a caller waiting for the metadata blocked forever; the error was never logged. The metadata is now completed with the stream error when the stream ends before it started. --- .../features/domain/FeatureStreamImpl.java | 27 ++++++- .../domain/FeatureStreamMetadataSpec.groovy | 78 +++++++++++++++++++ 2 files changed, 103 insertions(+), 2 deletions(-) create mode 100644 xtraplatform-features/src/test/groovy/de/ii/xtraplatform/features/domain/FeatureStreamMetadataSpec.groovy diff --git a/xtraplatform-features/src/main/java/de/ii/xtraplatform/features/domain/FeatureStreamImpl.java b/xtraplatform-features/src/main/java/de/ii/xtraplatform/features/domain/FeatureStreamImpl.java index d4f797a9f..9f770df30 100644 --- a/xtraplatform-features/src/main/java/de/ii/xtraplatform/features/domain/FeatureStreamImpl.java +++ b/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 CompletionStage completeMetadataOnEnd( + CompletionStage result, CompletableFuture 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 runWith( }); }; - return runner.runQuery(stream, query, mergedTransformations, !doTransform); + return completeMetadataOnEnd( + runner.runQuery(stream, query, mergedTransformations, !doTransform), onCollectionMetadata); } @Override @@ -336,7 +358,8 @@ public CompletionStage> runWith( }); }; - return runner.runQuery(stream, query, mergedTransformations, !doTransform); + return completeMetadataOnEnd( + runner.runQuery(stream, query, mergedTransformations, !doTransform), onCollectionMetadata); } private FeatureTokenSource getFeatureTokenSourceTransformed( diff --git a/xtraplatform-features/src/test/groovy/de/ii/xtraplatform/features/domain/FeatureStreamMetadataSpec.groovy b/xtraplatform-features/src/test/groovy/de/ii/xtraplatform/features/domain/FeatureStreamMetadataSpec.groovy new file mode 100644 index 000000000..0016ec3e5 --- /dev/null +++ b/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() + + 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. failedFuture(error) + def metadata = new CompletableFuture() + + 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() + + 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. 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) + } +}