Bugfix/8934 handle estimation sse cancel

.gitignore

@@ -5,3 +5,4 @@
 *.iml
 **/.git-versioned-pom.xml
 tmp/
+.mcp.json

server/pom.xml

@@ -65,6 +65,12 @@
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-tomcat</artifactId>
         </dependency>
+        <!-- WebClient for the streamed DAS endpoints. Not spring-boot-starter-webflux: that
+             drags in reactor-netty and a second event loop; the app stays servlet-based. -->
+        <dependency>
+            <groupId>org.springframework</groupId>
+            <artifactId>spring-webflux</artifactId>
+        </dependency>
         <!-- Spring Boot Cache Starter (includes spring-context) -->
         <dependency>
             <groupId>org.springframework.boot</groupId>

server/src/main/java/au/org/aodn/ogcapi/server/core/configuration/Config.java

@@ -1,5 +1,6 @@
 package au.org.aodn.ogcapi.server.core.configuration;
 
+import au.org.aodn.ogcapi.server.core.http.CancelPropagatingJdkConnector;
 import au.org.aodn.ogcapi.server.core.service.das.DasProperties;
 import au.org.aodn.ogcapi.server.core.service.dda.DdaProperties;
 import au.org.aodn.ogcapi.server.core.service.geonetwork.GNProperties;
@@ -16,9 +17,14 @@
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.http.HttpHeaders;
+import org.springframework.http.client.ClientHttpRequestInterceptor;
 import org.springframework.http.client.SimpleClientHttpRequestFactory;
+import org.springframework.http.codec.json.Jackson2JsonEncoder;
 import org.springframework.scheduling.annotation.EnableScheduling;
 import org.springframework.web.client.RestTemplate;
+import org.springframework.web.reactive.function.client.WebClient;
+
+import java.net.http.HttpClient;
 
 @Configuration
 @EnableScheduling
@@ -31,6 +37,7 @@
 public class Config {
 
     public static final String DAS_REST_TEMPLATE = "dasRestTemplate";
+    public static final String DAS_SSE_WEB_CLIENT = "dasSseWebClient";
 
     @Autowired
     ObjectMapper mapper;
@@ -69,15 +76,53 @@ public RestTemplate createDasRestTemplate(DasProperties dasProperties) {
         factory.setReadTimeout(dasProperties.readTimeout());
 
         RestTemplate restTemplate = new RestTemplate(factory);
-        restTemplate.getInterceptors().add((request, body, execution) -> {
+        restTemplate.getInterceptors().add(dasCredentials(dasProperties));
+        return restTemplate;
+    }
+
+    /**
+     * The DAS client for streamed endpoints (the cloud-optimised size estimate). A WebClient
+     * rather than a RestTemplate because a stream has to be cancellable: stopping the read is
+     * what abandons the estimate, and only a cancel reaches the socket. Two things to know:
+     * 1. The connector is customised so a cancel really does reach the socket, see
+     * CancelPropagatingJdkConnector.
+     * 2. There is no timeout here. DasService caps each frame gap with sseIdleTimeout instead.
+     */
+    @Bean(name = DAS_SSE_WEB_CLIENT, defaultCandidate = false)
+    public WebClient createDasSseWebClient(DasProperties dasProperties, ObjectMapper objectMapper) {
+        HttpClient httpClient = HttpClient.newBuilder()
+                .connectTimeout(dasProperties.connectTimeout())
+                // HttpURLConnection follows redirects on GET; the JDK client follows none by
+                // default, so ask for the equivalent rather than silently changing behaviour.
+                .followRedirects(HttpClient.Redirect.NORMAL)
+                .build();
+
+        WebClient.Builder builder = WebClient.builder()
+                .clientConnector(new CancelPropagatingJdkConnector(httpClient))
+                .baseUrl(dasProperties.host())
+                // The default codec builds its own ObjectMapper. Pass the application's so the DAS
+                // request body follows the same NON_NULL / JsonNullableModule config as everything else.
+                .codecs(codecs -> codecs.defaultCodecs().jackson2JsonEncoder(new Jackson2JsonEncoder(objectMapper)))
+                .defaultHeader("X-API-KEY", dasProperties.secret());
+
+        if (dasProperties.internal() != null) {
+            builder.defaultHeader("x-internal-das-header-secret", dasProperties.internal());
+        }
+        return builder.build();
+    }
+
+    /**
+     * Attaches the DAS credentials to every request.
+     */
+    private ClientHttpRequestInterceptor dasCredentials(DasProperties dasProperties) {
+        return (request, body, execution) -> {
             HttpHeaders headers = request.getHeaders();
             headers.set("X-API-KEY", dasProperties.secret());
             if (dasProperties.internal() != null) {
                 headers.set("x-internal-das-header-secret", dasProperties.internal());
             }
             return execution.execute(request, body);
-        });
-        return restTemplate;
+        };
     }
 
     @Bean

server/src/main/java/au/org/aodn/ogcapi/server/core/exception/SseClientGoneException.java

@@ -0,0 +1,31 @@
+package au.org.aodn.ogcapi.server.core.exception;
+
+import java.io.IOException;
+
+/**
+ * Raised when a write to an SSE client fails because the client has disconnected.
+ */
+public class SseClientGoneException extends IOException {
+
+    public SseClientGoneException(String contextId, Throwable cause) {
+        super("SSE client disconnected for " + contextId, cause);
+    }
+
+    /**
+     * Find this exception in {@code throwable}'s cause chain, or null if it is not there.
+     * A disconnect that unwound an upstream read always reaches the caller nested inside
+     * something else: {@code RestTemplate} wraps it in a {@code ResourceAccessException}, and
+     * the streamed DAS estimate wraps it in an {@code UncheckedIOException}.
+     */
+    public static SseClientGoneException find(Throwable throwable) {
+        for (Throwable current = throwable; current != null; current = current.getCause()) {
+            if (current instanceof SseClientGoneException clientGone) {
+                return clientGone;
+            }
+            if (current.getCause() == current) {
+                break;
+            }
+        }
+        return null;
+    }
+}

server/src/main/java/au/org/aodn/ogcapi/server/core/http/CancelPropagatingJdkConnector.java

@@ -0,0 +1,222 @@
+package au.org.aodn.ogcapi.server.core.http;
+
+import org.reactivestreams.Publisher;
+import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.DataBufferFactory;
+import org.springframework.core.io.buffer.DataBufferUtils;
+import org.springframework.core.io.buffer.DefaultDataBufferFactory;
+import org.springframework.http.HttpCookie;
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.HttpMethod;
+import org.springframework.http.HttpStatusCode;
+import org.springframework.http.client.reactive.AbstractClientHttpRequest;
+import org.springframework.http.client.reactive.AbstractClientHttpResponse;
+import org.springframework.http.client.reactive.ClientHttpConnector;
+import org.springframework.http.client.reactive.ClientHttpRequest;
+import org.springframework.http.client.reactive.ClientHttpResponse;
+import org.springframework.util.CollectionUtils;
+import org.springframework.util.LinkedCaseInsensitiveMap;
+import org.springframework.util.LinkedMultiValueMap;
+import org.springframework.util.MultiValueMap;
+import reactor.adapter.JdkFlowAdapter;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Flow;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+/**
+ * A copy of Spring's JdkClientHttpConnector without the cache(0) it puts on the response body.
+ * That operator never disconnects from its upstream, so a cancel from downstream never reaches
+ * Flow.Subscription.cancel() and the socket stays open while the server keeps producing a
+ * response nobody will read. Cancelling is the documented way to abort a JDK exchange (see
+ * BodySubscribers.ofPublisher), and the DAS estimate stream needs it: when the browser goes
+ * away, DAS only finds out when its socket closes.
+ * Two things to know:
+ * 1. Response cookies are not adapted. DAS sets none, and this is the one part of the original
+ *    the fork drops.
+ * 2. There is no request timeout. HttpRequest.Builder#timeout does not cover the body of a
+ *    streamed response (JDK-8258397), so callers use a Reactor timeout instead.
+ * All of this goes away once RestClient supports SSE upstream (spring-framework#35164).
+ */
+public class CancelPropagatingJdkConnector implements ClientHttpConnector {
+
+    private final HttpClient httpClient;
+
+    private final DataBufferFactory bufferFactory = DefaultDataBufferFactory.sharedInstance;
+
+    public CancelPropagatingJdkConnector(HttpClient httpClient) {
+        this.httpClient = httpClient;
+    }
+
+    @Override
+    public Mono<ClientHttpResponse> connect(HttpMethod method, URI uri,
+                                            Function<? super ClientHttpRequest, Mono<Void>> requestCallback) {
+
+        JdkRequest request = new JdkRequest(method, uri, bufferFactory);
+
+        return requestCallback.apply(request).then(Mono.defer(() -> {
+            HttpRequest nativeRequest = request.getNativeRequest();
+
+            CompletableFuture<HttpResponse<Flow.Publisher<List<ByteBuffer>>>> future =
+                    httpClient.sendAsync(nativeRequest, HttpResponse.BodyHandlers.ofPublisher());
+
+            return Mono.fromCompletionStage(future)
+                    .map(response -> new JdkResponse(response, bufferFactory));
+        }));
+    }
+
+    /**
+     * The request side, same as Spring's but with no timeout. Copied because Spring's is package
+     * private, so the response fork cannot reuse it.
+     */
+    private static final class JdkRequest extends AbstractClientHttpRequest {
+
+        private final HttpMethod method;
+        private final URI uri;
+        private final DataBufferFactory bufferFactory;
+        private final HttpRequest.Builder builder;
+
+        private JdkRequest(HttpMethod method, URI uri, DataBufferFactory bufferFactory) {
+            this.method = method;
+            this.uri = uri;
+            this.bufferFactory = bufferFactory;
+            this.builder = HttpRequest.newBuilder(uri);
+        }
+
+        @Override
+        public HttpMethod getMethod() {
+            return method;
+        }
+
+        @Override
+        public URI getURI() {
+            return uri;
+        }
+
+        @Override
+        public DataBufferFactory bufferFactory() {
+            return bufferFactory;
+        }
+
+        @Override
+        @SuppressWarnings("unchecked")
+        public <T> T getNativeRequest() {
+            return (T) builder.build();
+        }
+
+        @Override
+        protected void applyHeaders() {
+            for (Map.Entry<String, List<String>> entry : getHeaders().entrySet()) {
+                if (entry.getKey().equalsIgnoreCase(HttpHeaders.CONTENT_LENGTH)) {
+                    // The JDK restricts this header; the body publisher below carries the length.
+                    continue;
+                }
+                for (String value : entry.getValue()) {
+                    builder.header(entry.getKey(), value);
+                }
+            }
+            if (!getHeaders().containsKey(HttpHeaders.ACCEPT)) {
+                builder.header(HttpHeaders.ACCEPT, "*/*");
+            }
+        }
+
+        @Override
+        protected void applyCookies() {
+            MultiValueMap<String, HttpCookie> cookies = getCookies();
+            if (cookies.isEmpty()) {
+                return;
+            }
+            builder.header(HttpHeaders.COOKIE, cookies.values().stream()
+                    .flatMap(List::stream)
+                    .map(HttpCookie::toString)
+                    .collect(Collectors.joining(";")));
+        }
+
+        @Override
+        public Mono<Void> writeWith(Publisher<? extends DataBuffer> body) {
+            return doCommit(() -> {
+                builder.method(method.name(), toBodyPublisher(body));
+                return Mono.empty();
+            });
+        }
+
+        @Override
+        public Mono<Void> writeAndFlushWith(Publisher<? extends Publisher<? extends DataBuffer>> body) {
+            return writeWith(Flux.from(body).flatMap(Function.identity()));
+        }
+
+        @Override
+        public Mono<Void> setComplete() {
+            return doCommit(() -> {
+                builder.method(method.name(), HttpRequest.BodyPublishers.noBody());
+                return Mono.empty();
+            });
+        }
+
+        private HttpRequest.BodyPublisher toBodyPublisher(Publisher<? extends DataBuffer> body) {
+            Publisher<ByteBuffer> byteBuffers = body instanceof Mono ?
+                    Mono.from(body).map(JdkRequest::toByteBuffer) :
+                    Flux.from(body).map(JdkRequest::toByteBuffer);
+
+            Flow.Publisher<ByteBuffer> flow = JdkFlowAdapter.publisherToFlowPublisher(byteBuffers);
+            long contentLength = getHeaders().getContentLength();
+
+            return contentLength > 0 ?
+                    HttpRequest.BodyPublishers.fromPublisher(flow, contentLength) :
+                    HttpRequest.BodyPublishers.fromPublisher(flow);
+        }
+
+        private static ByteBuffer toByteBuffer(DataBuffer dataBuffer) {
+            ByteBuffer byteBuffer = ByteBuffer.allocate(dataBuffer.readableByteCount());
+            dataBuffer.toByteBuffer(byteBuffer);
+            return byteBuffer;
+        }
+    }
+
+    /**
+     * The response side. The missing cache(0) on the body is the whole reason this file exists.
+     */
+    private static final class JdkResponse extends AbstractClientHttpResponse {
+
+        private JdkResponse(HttpResponse<Flow.Publisher<List<ByteBuffer>>> response, DataBufferFactory bufferFactory) {
+            super(HttpStatusCode.valueOf(response.statusCode()),
+                    adaptHeaders(response),
+                    new LinkedMultiValueMap<>(),
+                    adaptBody(response, bufferFactory));
+        }
+
+        private static HttpHeaders adaptHeaders(HttpResponse<Flow.Publisher<List<ByteBuffer>>> response) {
+            Map<String, List<String>> rawHeaders = response.headers().map();
+            Map<String, List<String>> map = new LinkedCaseInsensitiveMap<>(rawHeaders.size(), Locale.ROOT);
+            MultiValueMap<String, String> multiValueMap = CollectionUtils.toMultiValueMap(map);
+            multiValueMap.putAll(rawHeaders);
+            return HttpHeaders.readOnlyHttpHeaders(multiValueMap);
+        }
+
+        private static Flux<DataBuffer> adaptBody(HttpResponse<Flow.Publisher<List<ByteBuffer>>> response,
+                                                  DataBufferFactory bufferFactory) {
+
+            Flow.Publisher<List<ByteBuffer>> body = response.body();
+            if (body == null) {
+                return Flux.empty();
+            }
+
+            // No cache(0) here: a cancel from downstream has to reach the JDK subscription.
+            return JdkFlowAdapter.flowPublisherToFlux(body)
+                    .flatMapIterable(Function.identity())
+                    .map(bufferFactory::wrap)
+                    .doOnDiscard(DataBuffer.class, DataBufferUtils::release);
+        }
+    }
+}

server/src/main/java/au/org/aodn/ogcapi/server/core/service/das/DasProperties.java

@@ -12,6 +12,7 @@ public record DasProperties(
         String secret,
         String internal,
         @DefaultValue("5s") Duration connectTimeout,
-        @DefaultValue("30s") Duration readTimeout
+        @DefaultValue("30s") Duration readTimeout,
+        @DefaultValue("2m") Duration sseIdleTimeout
 ) {
 }