diff --git a/.gitignore b/.gitignore index 70e74aca..83cef5c4 100644 --- a/.gitignore +++ b/.gitignore @@ -5,3 +5,4 @@ *.iml **/.git-versioned-pom.xml tmp/ +.mcp.json diff --git a/server/pom.xml b/server/pom.xml index 2f129b32..ab375258 100644 --- a/server/pom.xml +++ b/server/pom.xml @@ -65,6 +65,12 @@ org.springframework.boot spring-boot-starter-tomcat + + + org.springframework + spring-webflux + org.springframework.boot diff --git a/server/src/main/java/au/org/aodn/ogcapi/server/core/configuration/Config.java b/server/src/main/java/au/org/aodn/ogcapi/server/core/configuration/Config.java index 758a8089..7fa8bd1b 100644 --- a/server/src/main/java/au/org/aodn/ogcapi/server/core/configuration/Config.java +++ b/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 diff --git a/server/src/main/java/au/org/aodn/ogcapi/server/core/exception/SseClientGoneException.java b/server/src/main/java/au/org/aodn/ogcapi/server/core/exception/SseClientGoneException.java new file mode 100644 index 00000000..d0e4bc2b --- /dev/null +++ b/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; + } +} diff --git a/server/src/main/java/au/org/aodn/ogcapi/server/core/http/CancelPropagatingJdkConnector.java b/server/src/main/java/au/org/aodn/ogcapi/server/core/http/CancelPropagatingJdkConnector.java new file mode 100644 index 00000000..663abc5b --- /dev/null +++ b/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 connect(HttpMethod method, URI uri, + Function> requestCallback) { + + JdkRequest request = new JdkRequest(method, uri, bufferFactory); + + return requestCallback.apply(request).then(Mono.defer(() -> { + HttpRequest nativeRequest = request.getNativeRequest(); + + CompletableFuture>>> 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 getNativeRequest() { + return (T) builder.build(); + } + + @Override + protected void applyHeaders() { + for (Map.Entry> 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 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 writeWith(Publisher body) { + return doCommit(() -> { + builder.method(method.name(), toBodyPublisher(body)); + return Mono.empty(); + }); + } + + @Override + public Mono writeAndFlushWith(Publisher> body) { + return writeWith(Flux.from(body).flatMap(Function.identity())); + } + + @Override + public Mono setComplete() { + return doCommit(() -> { + builder.method(method.name(), HttpRequest.BodyPublishers.noBody()); + return Mono.empty(); + }); + } + + private HttpRequest.BodyPublisher toBodyPublisher(Publisher body) { + Publisher byteBuffers = body instanceof Mono ? + Mono.from(body).map(JdkRequest::toByteBuffer) : + Flux.from(body).map(JdkRequest::toByteBuffer); + + Flow.Publisher 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>> response, DataBufferFactory bufferFactory) { + super(HttpStatusCode.valueOf(response.statusCode()), + adaptHeaders(response), + new LinkedMultiValueMap<>(), + adaptBody(response, bufferFactory)); + } + + private static HttpHeaders adaptHeaders(HttpResponse>> response) { + Map> rawHeaders = response.headers().map(); + Map> map = new LinkedCaseInsensitiveMap<>(rawHeaders.size(), Locale.ROOT); + MultiValueMap multiValueMap = CollectionUtils.toMultiValueMap(map); + multiValueMap.putAll(rawHeaders); + return HttpHeaders.readOnlyHttpHeaders(multiValueMap); + } + + private static Flux adaptBody(HttpResponse>> response, + DataBufferFactory bufferFactory) { + + Flow.Publisher> 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); + } + } +} diff --git a/server/src/main/java/au/org/aodn/ogcapi/server/core/service/das/DasProperties.java b/server/src/main/java/au/org/aodn/ogcapi/server/core/service/das/DasProperties.java index fd0f8f9f..5c7ab0f1 100644 --- a/server/src/main/java/au/org/aodn/ogcapi/server/core/service/das/DasProperties.java +++ b/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 ) { } diff --git a/server/src/main/java/au/org/aodn/ogcapi/server/core/service/das/DasService.java b/server/src/main/java/au/org/aodn/ogcapi/server/core/service/das/DasService.java index a9042629..22c58026 100644 --- a/server/src/main/java/au/org/aodn/ogcapi/server/core/service/das/DasService.java +++ b/server/src/main/java/au/org/aodn/ogcapi/server/core/service/das/DasService.java @@ -3,42 +3,55 @@ import au.org.aodn.ogcapi.server.core.configuration.Config; import au.org.aodn.ogcapi.server.core.model.DatasetMetadata; import au.org.aodn.ogcapi.server.core.service.ApplicationInfo; -import au.org.aodn.ogcapi.server.core.util.SseResponseParser; +import au.org.aodn.ogcapi.server.core.util.DasSseFrames; import com.fasterxml.jackson.databind.ObjectMapper; import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.core.ParameterizedTypeReference; import org.springframework.http.*; +import org.springframework.http.codec.ServerSentEvent; import org.springframework.stereotype.Service; import org.springframework.web.client.RestTemplate; +import org.springframework.web.reactive.function.client.WebClient; import org.springframework.web.util.UriComponentsBuilder; +import reactor.core.publisher.Flux; +import java.io.IOException; +import java.io.UncheckedIOException; import java.util.Collections; import java.util.HashMap; -import java.util.List; +import java.util.Iterator; import java.util.Map; +import java.util.stream.Stream; @Service("DataAccessService") public class DasService implements ApplicationInfo { + private static final ParameterizedTypeReference> SSE_FRAME = + new ParameterizedTypeReference<>() { + }; + protected final DasProperties dasProperties; protected final RestTemplate httpClient; + protected final WebClient sseHttpClient; protected final ObjectMapper objectMapper; - protected final Map> appInfo; + protected final Map> appInfo; public DasService( DasProperties dasProperties, @Qualifier(Config.DAS_REST_TEMPLATE) RestTemplate httpClient, + @Qualifier(Config.DAS_SSE_WEB_CLIENT) WebClient sseHttpClient, ObjectMapper objectMapper) { this.dasProperties = dasProperties; this.httpClient = httpClient; + this.sseHttpClient = sseHttpClient; this.objectMapper = objectMapper; this.appInfo = queryInfo(httpClient, dasProperties.host(), dasProperties.infoPath()); } /** - * GET a feature-collection from the DAS, optionally bounded by start/end date. Only the date - * query params that are non-null are added, so a null value is never passed to URI template - * expansion (which would throw). Any path variables in {@code path} are supplied via - * {@code pathVariables}. + * GET a feature-collection from DAS, optionally bounded by start/end date. Only non-null dates + * are added as query params, because expanding a null URI template variable would throw. Any + * path variables in path come from pathVariables. */ private ResponseEntity getFeatureCollection(String path, String start, String end, Map pathVariables) { UriComponentsBuilder builder = UriComponentsBuilder.fromUriString(dasProperties.host() + path); @@ -90,34 +103,67 @@ public ResponseEntity getMooringDetailsBetweenDates(String startDateTime } /** - * Call the data-access-service cloud-optimised size estimate endpoint. - * The {@code parameters} map is the same batch-style subset request the - * download job submits (see {@code SubsetParametersUtils}), so DAS interprets - * the estimate and the download identically. Returns the estimate JSON so the - * SSE layer can forward it to the frontend unchanged. - *

- * DAS streams this endpoint over SSE: it heartbeats while computing and then - * sends the estimate in a terminal event, so the body is read to completion and - * unwrapped by {@link SseResponseParser}. Because the stream returns 200 as soon - * as it opens, a failed estimate arrives as an {@code error} event rather than an - * error status — the parser turns those back into an exception. Only failures - * raised before the stream starts (auth, API not ready) are still HTTP errors. + * Ask DAS for a cloud-optimised size estimate and return the estimate JSON unchanged, for the + * SSE layer to forward on. The parameters map is the same subset request the download job + * sends (see SubsetParametersUtils), so DAS treats both alike. Three things to know: + * 1. DAS answers over SSE: heartbeats while it computes, then the estimate as a final event. + * The stream returns 200 as soon as it opens, so a failed estimate arrives as an error event + * that DasSseFrames turns into an exception. Only failures before the stream opens (auth, API + * not ready) are HTTP errors, and onStatus rewrites those to hide the DAS host. + * 2. The response is not buffered. onHeartbeat runs on this thread for each heartbeat, and + * callers write to their own SSE client there, which is the only way to notice that client + * has gone. The IOException it throws unwinds this call, and leaving the try-with-resources + * cancels the Flux, closing the connection so DAS stops the estimate. That cancel only + * reaches the socket because of CancelPropagatingJdkConnector. + * 3. sseIdleTimeout is the gap allowed between frames, not a limit on the whole call. A slow + * estimate is fine while DAS keeps heartbeating; a silent DAS is given up on. */ - public String estimateCloudOptimisedDownloadSize(String uuid, Map parameters) { + public String estimateCloudOptimisedDownloadSize(String uuid, + Map parameters, + DasSseFrames.FrameCallback onHeartbeat) { - String url = UriComponentsBuilder.fromUriString(dasProperties.host() + "/api/v1/das/data/{uuid}/estimate_size") - .encode() - .toUriString(); - - Map uriVars = new HashMap<>(); - uriVars.put("uuid", uuid); + Flux> frames = sseHttpClient.post() + .uri("/api/v1/das/data/{uuid}/estimate_size", uuid) + .accept(MediaType.TEXT_EVENT_STREAM) + .contentType(MediaType.APPLICATION_JSON) + .bodyValue(parameters) + .retrieve() + .onStatus(HttpStatusCode::isError, response -> response.bodyToMono(String.class) + .defaultIfEmpty("") + .map(body -> new RuntimeException(describe(response.statusCode(), body)))) + .bodyToFlux(SSE_FRAME) + .timeout(dasProperties.sseIdleTimeout()); + + boolean sawFrame = false; + + // Closing the stream cancels the Flux, and that is what closes the connection to DAS. + try (Stream> stream = frames.toStream()) { + for (Iterator> it = stream.iterator(); it.hasNext(); ) { + sawFrame = true; + String payload = DasSseFrames.readTerminalFrame(objectMapper, it.next()); + if (payload != null) { + return payload; + } + onHeartbeat.onFrame(); + } + } catch (IOException e) { + throw new UncheckedIOException(e); + } - HttpHeaders headers = new HttpHeaders(); - headers.setAccept(List.of(MediaType.TEXT_EVENT_STREAM)); - headers.setContentType(MediaType.APPLICATION_JSON); + throw new RuntimeException(sawFrame ? + "data-access-service stream ended without a result or error event" : + "Empty response from data-access-service"); + } - String body = httpClient.postForObject(url, new HttpEntity<>(parameters, headers), String.class, uriVars); - return SseResponseParser.extractResultData(objectMapper, body); + /** + * Describe a failure that arrived before the stream opened. WebClient's own message quotes the + * request URL, which would show the DAS host to a user, so only the status and whatever DAS + * said are reported. + */ + private static String describe(HttpStatusCode status, String body) { + String reason = status instanceof HttpStatus known ? " " + known.getReasonPhrase() : ""; + String failure = "data-access-service returned " + status.value() + reason; + return body.isBlank() ? failure : failure + ": " + body; } public ResponseEntity getDatasetMetadata(String datasetId) { diff --git a/server/src/main/java/au/org/aodn/ogcapi/server/core/service/geoserver/wms/WmsServer.java b/server/src/main/java/au/org/aodn/ogcapi/server/core/service/geoserver/wms/WmsServer.java index 13e5de0b..38b628bc 100644 --- a/server/src/main/java/au/org/aodn/ogcapi/server/core/service/geoserver/wms/WmsServer.java +++ b/server/src/main/java/au/org/aodn/ogcapi/server/core/service/geoserver/wms/WmsServer.java @@ -707,7 +707,7 @@ public List fetchCapabilitiesLayersByUrl(String wmsServerUrl) { .getRootLayer() .getLayers(); - log.info("Fetched and cached get-capabilities layers {} ", layers); + log.info("Fetched and cached {} get-capabilities layers for {}", layers.size(), wmsServerUrl); return layers; } } diff --git a/server/src/main/java/au/org/aodn/ogcapi/server/core/exception/wfs/WfsErrorHandler.java b/server/src/main/java/au/org/aodn/ogcapi/server/core/service/sse/SseErrorHandler.java similarity index 85% rename from server/src/main/java/au/org/aodn/ogcapi/server/core/exception/wfs/WfsErrorHandler.java rename to server/src/main/java/au/org/aodn/ogcapi/server/core/service/sse/SseErrorHandler.java index 93b1dc53..f6809e41 100644 --- a/server/src/main/java/au/org/aodn/ogcapi/server/core/exception/wfs/WfsErrorHandler.java +++ b/server/src/main/java/au/org/aodn/ogcapi/server/core/service/sse/SseErrorHandler.java @@ -1,4 +1,4 @@ -package au.org.aodn.ogcapi.server.core.exception.wfs; +package au.org.aodn.ogcapi.server.core.service.sse; import au.org.aodn.ogcapi.server.core.exception.GeoserverFieldsNotFoundException; import au.org.aodn.ogcapi.server.core.exception.UnauthorizedServerException; @@ -13,12 +13,16 @@ import java.util.Set; import java.util.concurrent.ConcurrentHashMap; +/** + * Shared error handling for every SSE stream created by SseStreamHandler. It categorises the + * failure, runs the caller's cleanup, tells the client, and completes the emitter. + */ @Slf4j -public class WfsErrorHandler { +public class SseErrorHandler { private static final Set handledEmitters = Collections.newSetFromMap(new ConcurrentHashMap<>()); private enum ErrorType { - WFS_SERVER_ERROR, + UPSTREAM_SERVER_ERROR, NETWORK_ERROR, VALIDATION_ERROR, UNAUTHORIZED_SERVER_ERROR, @@ -45,14 +49,14 @@ public static void handleError(Exception e, String uuid, SseEmitter emitter, Run } switch (errorType) { - case WFS_SERVER_ERROR -> { - // ERROR level so New Relic can alert on failed downloads and identify + case UPSTREAM_SERVER_ERROR -> { + // ERROR level so New Relic can alert on failed streams and identify // the problematic upstream server (the URL is in the exception message). - log.error("WFS server error during download for UUID {}", uuid, e); + log.error("Upstream server error during SSE stream for UUID {}", uuid, e); emitter.send(SseEmitter.event() .name(SseEventName.ERROR.getValue()) .data(Map.of( - "message", "WFS server error", + "message", "Upstream server error", "timestamp", System.currentTimeMillis() ))); emitter.completeWithError(e); @@ -75,11 +79,11 @@ public static void handleError(Exception e, String uuid, SseEmitter emitter, Run } case UNAUTHORIZED_SERVER_ERROR -> { - log.warn("Unauthorized wfs server for UUID {}", uuid, e); + log.warn("Unauthorized upstream server for UUID {}", uuid, e); emitter.send(SseEmitter.event() .name(SseEventName.ERROR.getValue()) .data(Map.of( - "message", "Unauthorized wfs server", + "message", "Unauthorized upstream server", "timestamp", System.currentTimeMillis() ))); emitter.completeWithError(e); @@ -97,7 +101,7 @@ public static void handleError(Exception e, String uuid, SseEmitter emitter, Run } case UNKNOWN_ERROR -> { - log.error("Unknown error during WFS download for UUID {}", uuid, e); + log.error("Unknown error during SSE stream for UUID {}", uuid, e); emitter.send(SseEmitter.event() .name(SseEventName.ERROR.getValue()) .data(Map.of( @@ -118,11 +122,11 @@ public static void handleError(Exception e, String uuid, SseEmitter emitter, Run } private static ErrorType categorizeError(Exception e) { - // Upstream WFS server rejected the request (4xx/5xx). Must be checked before + // The upstream server rejected the request (4xx/5xx). Must be checked before // the network heuristics below so a failing server is never mistaken for a // client disconnect. if (e instanceof HttpStatusCodeException) { - return ErrorType.WFS_SERVER_ERROR; + return ErrorType.UPSTREAM_SERVER_ERROR; } if (e instanceof IOException || e instanceof IllegalStateException || e.getMessage() != null && ( @@ -137,7 +141,7 @@ private static ErrorType categorizeError(Exception e) { return ErrorType.VALIDATION_ERROR; } - // Unauthorized wfs error + // Unauthorized upstream server error if (e instanceof UnauthorizedServerException) { return ErrorType.UNAUTHORIZED_SERVER_ERROR; } diff --git a/server/src/main/java/au/org/aodn/ogcapi/server/core/service/sse/SseSession.java b/server/src/main/java/au/org/aodn/ogcapi/server/core/service/sse/SseSession.java index e2b52d06..8a4cd5ab 100644 --- a/server/src/main/java/au/org/aodn/ogcapi/server/core/service/sse/SseSession.java +++ b/server/src/main/java/au/org/aodn/ogcapi/server/core/service/sse/SseSession.java @@ -1,6 +1,6 @@ package au.org.aodn.ogcapi.server.core.service.sse; -import au.org.aodn.ogcapi.server.core.exception.wfs.WfsErrorHandler; +import au.org.aodn.ogcapi.server.core.exception.SseClientGoneException; import au.org.aodn.ogcapi.server.core.model.enumeration.SseEventName; import lombok.Getter; import lombok.extern.slf4j.Slf4j; @@ -44,6 +44,21 @@ public void send(SseEventName eventName, Object data) throws IOException { emitter.send(SseEmitter.event().name(eventName.getValue()).data(data)); } + /** + * Send a keep-alive and report a dead client as SseClientGoneException. + * This is how a stream checks its client is still there while waiting on an upstream server, + * since TCP says nothing about a peer that has gone until you write to it. Call it from the + * thread blocked upstream, so the exception unwinds that read and closes the connection + * instead of leaving a server computing a result for nobody. + */ + public void probeClient(Object data) throws SseClientGoneException { + try { + send(SseEventName.KEEP_ALIVE, data); + } catch (IOException e) { + throw new SseClientGoneException(contextId, e); + } + } + /** * Start sending a {@code keep-alive} event every {@code intervalSeconds}. The * payload is recomputed each tick by {@code payloadSupplier} so callers can @@ -55,7 +70,10 @@ public void startKeepAlive(long intervalSeconds, Supplier payloadSupplie try { send(SseEventName.KEEP_ALIVE, payloadSupplier.get()); } catch (Exception e) { - WfsErrorHandler.handleError(e, contextId, emitter, this::cleanup); + // This only ends the ticker and the emitter: a disconnect noticed here cannot + // unwind a thread blocked on an upstream socket, so the work itself should + // probe the client instead, see probeClient. + SseErrorHandler.handleError(e, contextId, emitter, this::cleanup); } }, intervalSeconds, intervalSeconds, TimeUnit.SECONDS); diff --git a/server/src/main/java/au/org/aodn/ogcapi/server/core/service/sse/SseStreamHandler.java b/server/src/main/java/au/org/aodn/ogcapi/server/core/service/sse/SseStreamHandler.java index d341871b..3f1dbbd0 100644 --- a/server/src/main/java/au/org/aodn/ogcapi/server/core/service/sse/SseStreamHandler.java +++ b/server/src/main/java/au/org/aodn/ogcapi/server/core/service/sse/SseStreamHandler.java @@ -1,10 +1,14 @@ package au.org.aodn.ogcapi.server.core.service.sse; -import au.org.aodn.ogcapi.server.core.exception.wfs.WfsErrorHandler; import lombok.extern.slf4j.Slf4j; +import org.springframework.scheduling.concurrent.CustomizableThreadFactory; import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; -import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.SynchronousQueue; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; /** * Shared scaffolding for the long-running SSE endpoints (WFS download / estimate, @@ -15,6 +19,25 @@ @Slf4j public class SseStreamHandler { + private static final int CORE_STREAMS = 4; + private static final int MAX_STREAMS = 64; + private static final long IDLE_THREAD_KEEP_ALIVE_SECONDS = 60L; + + /** + * Streams block on upstream sockets for minutes, so they get their own pool rather than + * ForkJoinPool.commonPool(), which runs a single thread on a 2-vCPU container and cannot + * see a blocking socket read, so one slow estimate made every other SSE request wait. + * The SynchronousQueue means a stream that cannot get a thread is rejected straight away + * rather than queueing behind work that may run for minutes. + */ + private static final ExecutorService STREAM_EXECUTOR = new ThreadPoolExecutor( + CORE_STREAMS, + MAX_STREAMS, + IDLE_THREAD_KEEP_ALIVE_SECONDS, + TimeUnit.SECONDS, + new SynchronousQueue<>(), + new CustomizableThreadFactory("sse-stream-")); + private SseStreamHandler() { } @@ -34,7 +57,7 @@ public interface SseWork { * A never-timing-out {@link SseEmitter} is created, lifecycle callbacks are * wired to clean up the keep-alive resources, and any exception from the work * (including validation errors thrown at the start) is routed through - * {@link WfsErrorHandler}. The work is responsible for completing the stream + * SseErrorHandler. The work is responsible for completing the stream * once its result has been sent. * * @param contextId identifier (e.g. uuid) used for logging and error handling @@ -56,15 +79,20 @@ public static SseEmitter stream(String contextId, SseWork work) { }); emitter.onError(throwable -> - WfsErrorHandler.handleError((Exception) throwable, contextId, emitter, session::cleanup)); + SseErrorHandler.handleError((Exception) throwable, contextId, emitter, session::cleanup)); - CompletableFuture.runAsync(() -> { - try { - work.run(session); - } catch (Exception e) { - WfsErrorHandler.handleError(e, contextId, emitter, session::cleanup); - } - }); + try { + STREAM_EXECUTOR.execute(() -> { + try { + work.run(session); + } catch (Exception e) { + SseErrorHandler.handleError(e, contextId, emitter, session::cleanup); + } + }); + } catch (RejectedExecutionException e) { + log.error("No SSE worker available for {}; {} streams already running", contextId, MAX_STREAMS); + SseErrorHandler.handleError(e, contextId, emitter, session::cleanup); + } return emitter; } diff --git a/server/src/main/java/au/org/aodn/ogcapi/server/core/util/DasSseFrames.java b/server/src/main/java/au/org/aodn/ogcapi/server/core/util/DasSseFrames.java new file mode 100644 index 00000000..ed0ae964 --- /dev/null +++ b/server/src/main/java/au/org/aodn/ogcapi/server/core/util/DasSseFrames.java @@ -0,0 +1,90 @@ +package au.org.aodn.ogcapi.server.core.util; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import org.springframework.http.codec.ServerSentEvent; + +import java.io.IOException; + +/** + * Reads the data-access-service Server-Sent Events contract. Splitting the stream into frames is + * the HTTP client's job; what is left here is what DAS means by a frame. + * DAS wraps long-running endpoints in its sse_it decorator, which heartbeats while it works and + * then returns the value in a single final frame: + * event: processing + * data: {"status":"processing","message":"Processing your request..."} + * event: result + * data: {"status":"completed","message":"Done","data": { ...the actual payload... }} + * Two things to know: + * 1. Anything the endpoint throws arrives as a final error frame instead, still on an HTTP 200 + * because the stream has already started, so a failed call shows up only by reading the frames. + * 2. This handles only that single-final-frame shape, not the chunked sse_wrapper responses DAS + * uses elsewhere, which emit many result frames to collect. + */ +public final class DasSseFrames { + + private static final String DATA_FIELD = "data"; + private static final String MESSAGE_FIELD = "message"; + + private static final String RESULT_EVENT = "result"; + private static final String ERROR_EVENT = "error"; + + private DasSseFrames() { + } + + /** + * Notified once per non-final frame, that is, once per DAS heartbeat. + * It may throw IOException on purpose: the caller forwards a keep-alive to its own SSE client + * here, and a broken pipe from that write is the only way to learn the client has gone. + * Letting it propagate aborts the read of the DAS stream and closes that connection, so DAS + * stops working on a result nobody will read. + */ + @FunctionalInterface + public interface FrameCallback { + + /** + * Does nothing. For callers with no client of their own to keep alive. + */ + FrameCallback IGNORE = () -> { + }; + + void onFrame() throws IOException; + } + + /** + * Interpret one frame. Returns a result frame's nested data payload as JSON, or null for a + * heartbeat the caller should skip. Throws RuntimeException for an error frame, or a result + * frame with no payload; an error keeps DAS's own message so the caller can forward it. + */ + public static String readTerminalFrame(ObjectMapper objectMapper, ServerSentEvent frame) { + String event = frame.event(); + if (!RESULT_EVENT.equals(event) && !ERROR_EVENT.equals(event)) { + return null; + } + + // A frame with no data line at all reads as an empty document, same as before. + String data = frame.data() == null ? "" : frame.data(); + JsonNode node; + try { + node = objectMapper.readTree(data); + } catch (Exception e) { + throw new RuntimeException( + String.format("Unreadable data-access-service %s event: %s", event, data), e); + } + + if (ERROR_EVENT.equals(event)) { + JsonNode message = node.get(MESSAGE_FIELD); + // Rethrow the reason verbatim: callers prefix it with their own context, and + // DAS already prefixes it with the status it would have returned. + throw new RuntimeException(message != null && !message.isNull() ? + message.asText() : + "data-access-service reported an error with no message"); + } + + JsonNode payload = node.get(DATA_FIELD); + if (payload == null || payload.isNull()) { + throw new RuntimeException("data-access-service result event carried no data: " + data); + } + return payload.toString(); + } +} diff --git a/server/src/main/java/au/org/aodn/ogcapi/server/core/util/SseResponseParser.java b/server/src/main/java/au/org/aodn/ogcapi/server/core/util/SseResponseParser.java deleted file mode 100644 index ad08feb9..00000000 --- a/server/src/main/java/au/org/aodn/ogcapi/server/core/util/SseResponseParser.java +++ /dev/null @@ -1,136 +0,0 @@ -package au.org.aodn.ogcapi.server.core.util; - -import com.fasterxml.jackson.databind.JsonNode; -import com.fasterxml.jackson.databind.ObjectMapper; - -/** - * Extracts the payload of a data-access-service Server-Sent Events response. - *

- * DAS wraps long-running endpoints in its {@code sse_it} decorator, which keeps the - * connection alive with {@code processing} heartbeats and then delivers the return - * value in a single terminal frame: - * 

- * event: processing
- * data: {"status":"processing","message":"Processing your request..."}
- *
- * event: result
- * data: {"status":"completed","message":"Done","data": { ...the actual payload... }}
- *
- * Anything the endpoint throws arrives as a terminal {@code error} frame instead — - * on an HTTP 200, because the stream has already started, so a failed call is - * only detectable by reading the frames. - *

- * This parser targets that single-terminal-frame shape. It is not suitable for the - * chunked {@code sse_wrapper} responses DAS uses elsewhere, which emit many - * {@code result} frames that all have to be collected. - */ -public final class SseResponseParser { - - private static final String EVENT_FIELD = "event"; - private static final String DATA_FIELD = "data"; - private static final String MESSAGE_FIELD = "message"; - - private static final String RESULT_EVENT = "result"; - private static final String ERROR_EVENT = "error"; - - private SseResponseParser() { - } - - /** - * Read an SSE body and return the payload nested under the terminal {@code result} - * frame's {@code data} field, serialized as JSON. - * - * @throws RuntimeException if the stream carries an {@code error} frame (with DAS's - * own message, unmodified, so the caller can forward it), or - * if it ends without a terminal frame - */ - public static String extractResultData(ObjectMapper objectMapper, String body) { - if (body == null || body.isBlank()) { - throw new RuntimeException("Empty response from data-access-service"); - } - - String event = null; - StringBuilder data = new StringBuilder(); - - for (String line : body.lines().toList()) { - if (line.isEmpty()) { - // Blank line terminates a frame. - String payload = readTerminalFrame(objectMapper, event, data.toString()); - if (payload != null) { - return payload; - } - event = null; - data.setLength(0); - continue; - } - if (line.startsWith(":")) { - // Comment line, per the SSE spec. - continue; - } - - int colon = line.indexOf(':'); - String field = colon < 0 ? line : line.substring(0, colon); - String value = colon < 0 ? "" : line.substring(colon + 1); - // A single leading space after the colon is part of the framing, not the value. - if (value.startsWith(" ")) { - value = value.substring(1); - } - - if (EVENT_FIELD.equals(field)) { - event = value; - } - else if (DATA_FIELD.equals(field)) { - if (!data.isEmpty()) { - data.append('\n'); - } - data.append(value); - } - } - - // The last frame may not be followed by a blank line. - String payload = readTerminalFrame(objectMapper, event, data.toString()); - if (payload != null) { - return payload; - } - - throw new RuntimeException("data-access-service stream ended without a result or error event"); - } - - /** - * Interpret one complete frame. - * - * @return the payload for a {@code result} frame, or null for any frame that is not - * terminal (heartbeats) and so should be skipped - * @throws RuntimeException for an {@code error} frame, or a {@code result} frame that - * carries no payload - */ - private static String readTerminalFrame(ObjectMapper objectMapper, String event, String data) { - if (!RESULT_EVENT.equals(event) && !ERROR_EVENT.equals(event)) { - return null; - } - - JsonNode node; - try { - node = objectMapper.readTree(data); - } - catch (Exception e) { - throw new RuntimeException( - String.format("Unreadable data-access-service %s event: %s", event, data), e); - } - - if (ERROR_EVENT.equals(event)) { - JsonNode message = node.get(MESSAGE_FIELD); - // Rethrow the reason verbatim: callers prefix it with their own context, and - // DAS already prefixes it with the status it would have returned. - throw new RuntimeException(message != null && !message.isNull() ? - message.asText() : - "data-access-service reported an error with no message"); - } - - JsonNode payload = node.get(DATA_FIELD); - if (payload == null || payload.isNull()) { - throw new RuntimeException("data-access-service result event carried no data: " + data); - } - return payload.toString(); - } -} diff --git a/server/src/main/java/au/org/aodn/ogcapi/server/processes/RestServices.java b/server/src/main/java/au/org/aodn/ogcapi/server/processes/RestServices.java index b91f163c..6fd7c363 100644 --- a/server/src/main/java/au/org/aodn/ogcapi/server/processes/RestServices.java +++ b/server/src/main/java/au/org/aodn/ogcapi/server/processes/RestServices.java @@ -1,5 +1,6 @@ package au.org.aodn.ogcapi.server.processes; +import au.org.aodn.ogcapi.server.core.exception.SseClientGoneException; import au.org.aodn.ogcapi.server.core.model.enumeration.DatasetDownloadEnums; import au.org.aodn.ogcapi.server.core.model.enumeration.SseEventName; import au.org.aodn.ogcapi.server.core.model.ogc.FeatureRequest; @@ -331,17 +332,25 @@ public SseEmitter estimateCloudOptimisedDownloadWithSse(String uuid, "timestamp", System.currentTimeMillis() )); - // STEP 2: Start keep-alive mechanism while data-access-service computes the estimate - session.startKeepAlive(20, () -> Map.of( - "message", "Estimating download size...", - "timestamp", System.currentTimeMillis() - )); - - // STEP 3: Call the data-access-service estimate endpoint and forward the result + // STEP 2: Call the data-access-service estimate endpoint and forward the result. + // DAS's own heartbeats (~20s) drive the keep-alive instead of a timer thread, so a + // disconnected client breaks the thread out of the read, closing the connection to + // DAS and stopping an estimate nobody waits for. The frontend sees what it saw before. try { - String estimateJson = dasService.estimateCloudOptimisedDownloadSize(uuid, parameters); + String estimateJson = dasService.estimateCloudOptimisedDownloadSize(uuid, parameters, + () -> session.probeClient(Map.of( + "message", "Estimating download size...", + "timestamp", System.currentTimeMillis() + ))); session.send(SseEventName.ESTIMATE_COMPLETE, estimateJson); } catch (Exception e) { + SseClientGoneException clientGone = SseClientGoneException.find(e); + if (clientGone != null) { + // Not an estimate failure: the client left, and that is what aborted the + // call. Rethrow so the shared handler logs it as a disconnect; there is no + // socket left to report anything on. + throw clientGone; + } log.warn("Cloud-optimised size estimation failed for UUID {}: {}", uuid, e.getMessage()); session.send(SseEventName.ESTIMATE_FAILED, Map.of( "message", "Size estimation failed: " + e.getMessage(), diff --git a/server/src/main/resources/application.yaml b/server/src/main/resources/application.yaml index c11293f3..0d7ceb04 100644 --- a/server/src/main/resources/application.yaml +++ b/server/src/main/resources/application.yaml @@ -486,6 +486,7 @@ data-access-service: secret: 123 connect-timeout: 5s read-timeout: 30s + sse-idle-timeout: 2m management: info: diff --git a/server/src/test/java/au/org/aodn/ogcapi/server/core/configuration/ConfigTest.java b/server/src/test/java/au/org/aodn/ogcapi/server/core/configuration/ConfigTest.java index 3b49f452..6b384e4b 100644 --- a/server/src/test/java/au/org/aodn/ogcapi/server/core/configuration/ConfigTest.java +++ b/server/src/test/java/au/org/aodn/ogcapi/server/core/configuration/ConfigTest.java @@ -1,6 +1,8 @@ package au.org.aodn.ogcapi.server.core.configuration; +import au.org.aodn.ogcapi.server.core.http.RecordingSseConnector; import au.org.aodn.ogcapi.server.core.service.das.DasProperties; +import com.fasterxml.jackson.databind.ObjectMapper; import org.junit.jupiter.api.Test; import org.springframework.http.HttpHeaders; import org.springframework.http.HttpMethod; @@ -14,26 +16,27 @@ import java.net.URI; import java.time.Duration; +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; /** - * The DAS credentials are attached by the RestTemplate bean itself rather than by a per-call - * HttpEntity, so this is where that wiring is covered — including the fact that the template - * GeoServer shares must never carry them. + * The DAS credentials are attached by the client beans themselves, not per call, so this is where + * that wiring is covered, including the template GeoServer shares never carrying them. */ public class ConfigTest { private static final DasProperties DAS_PROPERTIES = new DasProperties( - "http://localhost:5000", null,"test-secret", "internal-secret", - Duration.ofSeconds(5), Duration.ofSeconds(30)); + "http://localhost:5000", null, "test-secret", "internal-secret", + Duration.ofSeconds(5), Duration.ofSeconds(30), Duration.ofMinutes(2)); private final Config config = new Config(); /** - * Runs a template's interceptor chain over a bare request and hands back the headers they set. - * Each interceptor only mutates headers before delegating, so a stubbed execution is enough. + * Runs a template's interceptor chain over a bare request and returns the headers they set. + * Interceptors only touch headers before delegating, so a stubbed execution is enough. */ private static HttpHeaders headersAfterInterceptors(RestTemplate template) throws IOException { MockClientHttpRequest request = new MockClientHttpRequest(HttpMethod.GET, URI.create("http://localhost:5000/probe")); @@ -54,6 +57,27 @@ public void testDasTemplateAttachesApiKeyButNotAccept() throws IOException { "the same client fetches JSON and binary tiles, so Accept is left to each call"); } + /** + * A WebClient has no interceptors to inspect, so the credentials are read back off a request + * it actually made, through a connector that records one. + */ + @Test + public void testDasSseClientAttachesTheSameCredentials() { + RecordingSseConnector connector = new RecordingSseConnector(); + config.createDasSseWebClient(DAS_PROPERTIES, new ObjectMapper()) + .mutate() + .clientConnector(connector) + .build() + .get() + .uri("/probe") + .retrieve() + .bodyToMono(String.class) + .block(); + + assertEquals("test-secret", connector.headers().getFirst("X-API-KEY")); + assertEquals("internal-secret", connector.headers().getFirst("x-internal-das-header-secret")); + } + @Test public void testApplicationWideTemplateCarriesNoDasCredentials() throws IOException { RestTemplate restTemplate = config.createRestTemplate(); @@ -68,8 +92,8 @@ public void testApplicationWideTemplateCarriesNoDasCredentials() throws IOExcept @Test public void testInternalSecretOmittedWhenNotConfigured() throws IOException { DasProperties noInternal = new DasProperties( - "http://localhost:5000", null,"test-secret", null, - Duration.ofSeconds(5), Duration.ofSeconds(30)); + "http://localhost:5000", null, "test-secret", null, + Duration.ofSeconds(5), Duration.ofSeconds(30), Duration.ofMinutes(2)); HttpHeaders headers = headersAfterInterceptors(config.createDasRestTemplate(noInternal)); diff --git a/server/src/test/java/au/org/aodn/ogcapi/server/core/http/RecordingSseConnector.java b/server/src/test/java/au/org/aodn/ogcapi/server/core/http/RecordingSseConnector.java new file mode 100644 index 00000000..307ab71a --- /dev/null +++ b/server/src/test/java/au/org/aodn/ogcapi/server/core/http/RecordingSseConnector.java @@ -0,0 +1,109 @@ +package au.org.aodn.ogcapi.server.core.http; + +import org.springframework.core.io.buffer.DataBufferFactory; +import org.springframework.core.io.buffer.DefaultDataBufferFactory; +import org.springframework.http.HttpHeaders; +import org.springframework.http.HttpMethod; +import org.springframework.http.HttpStatus; +import org.springframework.http.HttpStatusCode; +import org.springframework.http.MediaType; +import org.springframework.http.client.reactive.ClientHttpConnector; +import org.springframework.http.client.reactive.ClientHttpRequest; +import org.springframework.http.client.reactive.ClientHttpResponse; +import org.springframework.mock.http.client.reactive.MockClientHttpRequest; +import org.springframework.mock.http.client.reactive.MockClientHttpResponse; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + +import java.net.URI; +import java.nio.charset.StandardCharsets; +import java.util.List; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.Function; + +/** + * Hands a WebClient a canned SSE response and keeps what it was asked to send. + * A WebClient has no interceptors and no MockRestServiceServer, so the connector is the only place + * a test can see the request. This stands in for one and hands back frames one at a time, so a + * test can also tell whether the body was cancelled or read to the end. + */ +public final class RecordingSseConnector implements ClientHttpConnector { + + private static final DataBufferFactory BUFFERS = DefaultDataBufferFactory.sharedInstance; + + private HttpStatusCode status = HttpStatus.OK; + private List frames = List.of(); + private MockClientHttpRequest lastRequest; + private String lastBody; + private final AtomicBoolean cancelled = new AtomicBoolean(); + private final AtomicInteger framesDelivered = new AtomicInteger(); + + /** + * Answer with 200 and these SSE frames, each one a complete frame ending in a blank line. + */ + public RecordingSseConnector respondWith(List frames) { + this.status = HttpStatus.OK; + this.frames = frames; + return this; + } + + /** + * Answer with a status and no body, the way DAS fails before a stream opens. + */ + public RecordingSseConnector respondWith(HttpStatusCode status) { + this.status = status; + this.frames = List.of(); + return this; + } + + public HttpMethod method() { + return lastRequest.getMethod(); + } + + public URI uri() { + return lastRequest.getURI(); + } + + public HttpHeaders headers() { + return lastRequest.getHeaders(); + } + + public String body() { + return lastBody; + } + + public boolean wasCancelled() { + return cancelled.get(); + } + + public int framesDelivered() { + return framesDelivered.get(); + } + + @Override + public Mono connect(HttpMethod method, URI uri, + Function> requestCallback) { + + MockClientHttpRequest request = new MockClientHttpRequest(method, uri); + lastRequest = request; + + return requestCallback.apply(request) + // Deferred: the mock only has a body once the callback above has written one. + .then(Mono.defer(request::getBodyAsString)) + .doOnNext(body -> lastBody = body) + .then(Mono.fromSupplier(this::response)); + } + + private ClientHttpResponse response() { + MockClientHttpResponse response = new MockClientHttpResponse(status); + response.getHeaders().setContentType(MediaType.TEXT_EVENT_STREAM); + // One buffer per frame, handed over only as the reader asks for it, so a cancel shows + // up as frames that were never delivered. + response.setBody(Flux.fromIterable(frames) + .doOnNext(frame -> framesDelivered.incrementAndGet()) + .map(frame -> BUFFERS.wrap(frame.getBytes(StandardCharsets.UTF_8))) + .doOnCancel(() -> cancelled.set(true))); + return response; + } +} diff --git a/server/src/test/java/au/org/aodn/ogcapi/server/core/service/das/DasServiceEstimateStreamTest.java b/server/src/test/java/au/org/aodn/ogcapi/server/core/service/das/DasServiceEstimateStreamTest.java new file mode 100644 index 00000000..9e4ed3e1 --- /dev/null +++ b/server/src/test/java/au/org/aodn/ogcapi/server/core/service/das/DasServiceEstimateStreamTest.java @@ -0,0 +1,469 @@ +package au.org.aodn.ogcapi.server.core.service.das; + +import au.org.aodn.ogcapi.server.core.configuration.Config; +import au.org.aodn.ogcapi.server.core.http.RecordingSseConnector; +import au.org.aodn.ogcapi.server.core.util.DasSseFrames; +import com.fasterxml.jackson.databind.ObjectMapper; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; +import org.springframework.http.HttpHeaders; +import org.springframework.http.HttpMethod; +import org.springframework.http.HttpStatus; +import org.springframework.http.MediaType; +import org.springframework.http.client.reactive.ClientHttpConnector; +import org.springframework.web.client.RestTemplate; +import org.springframework.web.reactive.function.client.WebClient; + +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.io.UncheckedIOException; +import java.net.ServerSocket; +import java.net.Socket; +import java.net.SocketTimeoutException; +import java.net.URI; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Covers the streamed cloud-optimised size estimate: the request DAS receives, the heartbeats + * handed back as they arrive, and what happens when the caller's own client disappears + * mid-estimate. A mocked WebClient cannot show any of that, so these tests drive a real one: a + * stub connector for the frame-level assertions, and a real socket for the ones that only mean + * something on the wire, that the request body arrives and that the connection to DAS is dropped. + */ +public class DasServiceEstimateStreamTest { + + private static final String HOST = "http://localhost:5000"; + + private static final String HEARTBEAT_FRAME = """ + event: processing + data: {"status":"processing","message":"Processing your request..."} + + """; + + private static final String RESULT_FRAME = """ + event: result + data: {"status":"completed","data":{"estimated_output_bytes":123}} + + """; + + private RecordingSseConnector connector; + private DasService dasService; + + private static DasProperties properties(String host, Duration sseIdleTimeout) { + return new DasProperties(host, null, "test-secret", null, + Duration.ofSeconds(5), Duration.ofSeconds(30), sseIdleTimeout); + } + + private static DasService serviceOn(DasProperties properties, ClientHttpConnector connector) { + WebClient webClient = WebClient.builder() + .clientConnector(connector) + .baseUrl(properties.host()) + .build(); + return new DasService(properties, new RestTemplate(), webClient, new ObjectMapper()); + } + + /** + * Builds the service the way the application does, so the real connector is under test. + */ + private static DasService realServiceOn(DasProperties properties) { + return new DasService(properties, new RestTemplate(), + new Config().createDasSseWebClient(properties, new ObjectMapper()), new ObjectMapper()); + } + + @BeforeEach + public void setUp() { + connector = new RecordingSseConnector(); + dasService = serviceOn(properties(HOST, Duration.ofMinutes(2)), connector); + } + + private String estimate(DasSseFrames.FrameCallback onHeartbeat) { + return estimate(dasService, onHeartbeat); + } + + private static String estimate(DasService service, DasSseFrames.FrameCallback onHeartbeat) { + return service.estimateCloudOptimisedDownloadSize( + "test-uuid", + Map.of("uuid", "test-uuid", "key", "a.zarr", "output_format", "netcdf"), + onHeartbeat); + } + + @Test + public void testEstimateStreamsFramesAndUnwrapsTheResult() { + connector.respondWith(List.of(HEARTBEAT_FRAME, HEARTBEAT_FRAME, RESULT_FRAME)); + + AtomicInteger heartbeats = new AtomicInteger(); + String result = estimate(heartbeats::incrementAndGet); + + assertEquals("{\"estimated_output_bytes\":123}", result); + assertEquals(2, heartbeats.get(), "Each DAS heartbeat is handed to the caller as it arrives"); + } + + @Test + public void testEstimatePostsBatchStyleParametersAsJsonEventStream() { + connector.respondWith(List.of(RESULT_FRAME)); + + estimate(DasSseFrames.FrameCallback.IGNORE); + + assertEquals(HttpMethod.POST, connector.method()); + assertEquals(URI.create(HOST + "/api/v1/das/data/test-uuid/estimate_size"), connector.uri()); + assertEquals(MediaType.TEXT_EVENT_STREAM_VALUE, connector.headers().getFirst(HttpHeaders.ACCEPT)); + assertEquals(MediaType.APPLICATION_JSON_VALUE, connector.headers().getFirst(HttpHeaders.CONTENT_TYPE)); + + String sent = connector.body(); + assertTrue(sent.contains("\"uuid\":\"test-uuid\""), "Got: " + sent); + assertTrue(sent.contains("\"key\":\"a.zarr\""), "The batch-style parameters go to DAS unchanged: " + sent); + assertTrue(sent.contains("\"output_format\":\"netcdf\""), "Got: " + sent); + } + + /** + * The body has to survive the connector, not just WebClient. A connector that forgets to + * build a body publisher still opens the stream and still gets a 200, so nothing above this + * level would notice DAS being sent an empty estimate request. + */ + @Test + @Timeout(30) + public void testEstimateBodyReachesDasOverARealSocket() throws Exception { + try (ServerSocket serverSocket = new ServerSocket(0)) { + serverSocket.setSoTimeout(20_000); + CompletableFuture request = serveOneRequest(serverSocket, RESULT_FRAME); + + DasProperties properties = properties("http://localhost:" + serverSocket.getLocalPort(), Duration.ofSeconds(20)); + String result = estimate(realServiceOn(properties), DasSseFrames.FrameCallback.IGNORE); + + assertEquals("{\"estimated_output_bytes\":123}", result); + + String sent = request.get(20, TimeUnit.SECONDS); + assertTrue(sent.startsWith("POST /api/v1/das/data/test-uuid/estimate_size "), "Got: " + sent); + assertTrue(sent.contains("\"uuid\":\"test-uuid\""), "The request body must reach DAS: " + sent); + assertTrue(sent.contains("\"key\":\"a.zarr\""), "Got: " + sent); + assertTrue(sent.contains("\"output_format\":\"netcdf\""), "Got: " + sent); + } + } + + @Test + public void testFailedWriteToTheClientAbandonsTheStreamMidBody() { + // The client disconnects while DAS is still heartbeating: the write in the callback + // throws, and that must abort the read instead of running the estimate to completion. + List frames = new ArrayList<>(Collections.nCopies(500, HEARTBEAT_FRAME)); + frames.add(RESULT_FRAME); + connector.respondWith(frames); + + AtomicInteger heartbeats = new AtomicInteger(); + UncheckedIOException e = assertThrows(UncheckedIOException.class, () -> estimate(() -> { + if (heartbeats.incrementAndGet() == 2) { + throw new IOException("Broken pipe"); + } + })); + + assertEquals("Broken pipe", e.getCause().getMessage()); + assertEquals(2, heartbeats.get(), "The read stops at the failed write"); + assertTrue(connector.wasCancelled(), "The response body must be cancelled, not left running"); + assertTrue(connector.framesDelivered() < frames.size(), + "The body should be abandoned mid-stream, not drained to the end first"); + } + + @Test + public void testErrorFrameStillSurfacesAsAnException() { + // A failure raised after the stream opened arrives on an HTTP 200, so it is only visible + // in the frames — the SSE layer must not report it as a successful estimate. + connector.respondWith(List.of(HEARTBEAT_FRAME, """ + event: error + data: {"status":"error","message":"404: No matching keys found for uuid=test-uuid"} + + """)); + + RuntimeException e = assertThrows(RuntimeException.class, + () -> estimate(DasSseFrames.FrameCallback.IGNORE)); + + assertEquals("404: No matching keys found for uuid=test-uuid", e.getMessage(), + "DAS's reason is forwarded verbatim for the SSE layer to report"); + } + + @Test + public void testNon2xxPropagatesBeforeAnyFrameIsRead() { + // Failures raised before the stream starts (auth, API not ready) are still HTTP errors. + connector.respondWith(HttpStatus.NOT_FOUND); + + AtomicInteger heartbeats = new AtomicInteger(); + RuntimeException e = assertThrows(RuntimeException.class, () -> estimate(heartbeats::incrementAndGet)); + + assertEquals(0, heartbeats.get()); + assertTrue(e.getMessage().contains("404"), "Got: " + e.getMessage()); + assertFalse(e.getMessage().contains(HOST), + "WebClient quotes the request URL in its own message; the frontend shows this one: " + + e.getMessage()); + } + + /** + * At socket level: a real DAS-shaped server that keeps + * heartbeating, a client that goes away, and the assertion that the connection is dropped + * while the server still thinks it has work to do. Left unclosed, DAS would keep computing + * an estimate nobody will read. + */ + @Test + @Timeout(30) + public void testUpstreamSocketIsClosedWhenTheClientGoesAway() throws Exception { + try (ServerSocket serverSocket = new ServerSocket(0)) { + serverSocket.setSoTimeout(20_000); + CompletableFuture sawDisconnect = serveHeartbeatsUntilClientLeaves(serverSocket, 0); + + DasProperties properties = properties("http://localhost:" + serverSocket.getLocalPort(), Duration.ofMinutes(20)); + DasService service = realServiceOn(properties); + + AtomicInteger heartbeats = new AtomicInteger(); + assertThrows(UncheckedIOException.class, () -> estimate(service, () -> { + if (heartbeats.incrementAndGet() == 2) { + throw new IOException("Broken pipe"); + } + })); + + assertTrue(sawDisconnect.get(20, TimeUnit.SECONDS), + "The server must see the connection close while it is still heartbeating"); + } + } + + /** + * sseIdleTimeout is a gap between frames. A DAS that opens the stream and then says nothing + * is given up on, and the connection is closed rather than left pinning a worker thread. + */ + @Test + @Timeout(30) + public void testIdleTimeoutGivesUpOnASilentDas() throws Exception { + try (ServerSocket serverSocket = new ServerSocket(0)) { + serverSocket.setSoTimeout(20_000); + CompletableFuture sawDisconnect = serveHeadersThenSilence(serverSocket); + + DasProperties properties = properties("http://localhost:" + serverSocket.getLocalPort(), Duration.ofSeconds(1)); + DasService service = realServiceOn(properties); + + AtomicInteger heartbeats = new AtomicInteger(); + RuntimeException e = assertThrows(RuntimeException.class, + () -> estimate(service, heartbeats::incrementAndGet)); + + assertEquals(0, heartbeats.get()); + assertNotNull(causeOfType(e, TimeoutException.class), + "A silent DAS should time out, got: " + e); + assertTrue(sawDisconnect.get(20, TimeUnit.SECONDS), + "The abandoned stream must be closed, not left open"); + } + } + + /** + * The other half of the same semantics, and the regression the rename could have caused: a + * DAS that keeps heartbeating is never cut off, however long it takes. This used to be a + * ceiling on the whole exchange, which would kill an estimate that was working fine. + */ + @Test + @Timeout(30) + public void testIdleTimeoutDoesNotCapAStreamThatKeepsHeartbeating() throws Exception { + try (ServerSocket serverSocket = new ServerSocket(0)) { + serverSocket.setSoTimeout(20_000); + CompletableFuture sawDisconnect = serveHeartbeatsUntilClientLeaves(serverSocket, 100); + + DasProperties properties = properties("http://localhost:" + serverSocket.getLocalPort(), Duration.ofSeconds(1)); + DasService service = realServiceOn(properties); + + // Twenty heartbeats at 100ms apart is twice the idle timeout, so a whole-exchange + // cap would have fired well before the client stops of its own accord. + AtomicInteger heartbeats = new AtomicInteger(); + assertThrows(UncheckedIOException.class, () -> estimate(service, () -> { + if (heartbeats.incrementAndGet() == 20) { + throw new IOException("Broken pipe"); + } + })); + + assertEquals(20, heartbeats.get(), "A heartbeating stream must not be timed out"); + assertTrue(sawDisconnect.get(20, TimeUnit.SECONDS), + "The server must see the connection close"); + } + } + + private static T causeOfType(Throwable throwable, Class type) { + for (Throwable current = throwable; current != null; current = current.getCause()) { + if (type.isInstance(current)) { + return type.cast(current); + } + if (current.getCause() == current) { + break; + } + } + return null; + } + + /** + * Answers one request with an endless SSE heartbeat stream — DAS mid-estimate — and completes + * with true as soon as the client's end of the connection goes away. + */ + private CompletableFuture serveHeartbeatsUntilClientLeaves(ServerSocket serverSocket, long pauseMillis) { + CompletableFuture sawDisconnect = new CompletableFuture<>(); + + Thread server = new Thread(() -> { + try (Socket socket = serverSocket.accept()) { + socket.setSoTimeout(200); + InputStream in = socket.getInputStream(); + OutputStream out = socket.getOutputStream(); + + writeSseHeaders(out); + + // Keep heartbeating like a long estimate would, watching for the client to leave. + for (int i = 0; i < 400; i++) { + writeChunk(out, HEARTBEAT_FRAME); + if (clientHasGone(in)) { + sawDisconnect.complete(true); + return; + } + if (pauseMillis > 0) { + Thread.sleep(pauseMillis); + } + } + sawDisconnect.complete(false); + } catch (IOException e) { + // A write failing because the peer has gone is the same news, by a shorter route. + sawDisconnect.complete(true); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + }, "sse-test-server"); + + server.setDaemon(true); + server.start(); + return sawDisconnect; + } + + /** + * Opens the stream and then says nothing at all — a DAS that has stopped talking to us. + * Completes with true once the client gives up and closes the connection. + */ + private CompletableFuture serveHeadersThenSilence(ServerSocket serverSocket) { + CompletableFuture sawDisconnect = new CompletableFuture<>(); + + Thread server = new Thread(() -> { + try (Socket socket = serverSocket.accept()) { + socket.setSoTimeout(200); + writeSseHeaders(socket.getOutputStream()); + + // The first read hands back the request itself, so keep watching until the + // client actually goes rather than reading once and calling it a day. + InputStream in = socket.getInputStream(); + long deadline = System.currentTimeMillis() + 20_000; + while (System.currentTimeMillis() < deadline) { + if (clientHasGone(in)) { + sawDisconnect.complete(true); + return; + } + } + sawDisconnect.complete(false); + } catch (IOException e) { + sawDisconnect.complete(true); + } + }, "sse-silent-test-server"); + + server.setDaemon(true); + server.start(); + return sawDisconnect; + } + + /** + * Reads one whole request — head and body — and answers it with sseBody, so a test can + * assert on what actually went down the socket. + */ + private CompletableFuture serveOneRequest(ServerSocket serverSocket, String sseBody) { + CompletableFuture received = new CompletableFuture<>(); + + Thread server = new Thread(() -> { + try (Socket socket = serverSocket.accept()) { + socket.setSoTimeout(20_000); + String request = readRequest(socket.getInputStream()); + + OutputStream out = socket.getOutputStream(); + writeSseHeaders(out); + writeChunk(out, sseBody); + out.write("0\r\n\r\n".getBytes(StandardCharsets.US_ASCII)); + out.flush(); + + received.complete(request); + } catch (IOException e) { + received.completeExceptionally(e); + } + }, "sse-echo-test-server"); + + server.setDaemon(true); + server.start(); + return received; + } + + /** + * Reads the request head, then exactly as many body bytes as it declared. + */ + private static String readRequest(InputStream in) throws IOException { + ByteArrayOutputStream head = new ByteArrayOutputStream(); + int b; + while (!endsWithBlankLine(head) && (b = in.read()) != -1) { + head.write(b); + } + + String text = head.toString(StandardCharsets.UTF_8); + int contentLength = 0; + for (String line : text.split("\r\n")) { + if (line.toLowerCase(Locale.ROOT).startsWith("content-length:")) { + contentLength = Integer.parseInt(line.substring(line.indexOf(':') + 1).trim()); + } + } + return text + new String(in.readNBytes(contentLength), StandardCharsets.UTF_8); + } + + private static boolean endsWithBlankLine(ByteArrayOutputStream buffer) { + byte[] bytes = buffer.toByteArray(); + return bytes.length >= 4 + && bytes[bytes.length - 4] == '\r' && bytes[bytes.length - 3] == '\n' + && bytes[bytes.length - 2] == '\r' && bytes[bytes.length - 1] == '\n'; + } + + /** + * Reads whatever the client has sent (its request, then nothing) to find out whether the + * connection is still open. Reaching end of stream means it is not. + */ + private static boolean clientHasGone(InputStream in) throws IOException { + try { + return in.read(new byte[8192]) == -1; + } catch (SocketTimeoutException stillConnected) { + return false; + } + } + + private static void writeSseHeaders(OutputStream out) throws IOException { + out.write(("HTTP/1.1 200 OK\r\n" + + "Content-Type: text/event-stream\r\n" + + "Transfer-Encoding: chunked\r\n" + + "\r\n").getBytes(StandardCharsets.US_ASCII)); + out.flush(); + } + + private static void writeChunk(OutputStream out, String payload) throws IOException { + byte[] bytes = payload.getBytes(StandardCharsets.UTF_8); + out.write((Integer.toHexString(bytes.length) + "\r\n").getBytes(StandardCharsets.US_ASCII)); + out.write(bytes); + out.write("\r\n".getBytes(StandardCharsets.US_ASCII)); + out.flush(); + } + +} diff --git a/server/src/test/java/au/org/aodn/ogcapi/server/core/service/das/DasServiceHeadersTest.java b/server/src/test/java/au/org/aodn/ogcapi/server/core/service/das/DasServiceHeadersTest.java index f5b56170..757ad1cf 100644 --- a/server/src/test/java/au/org/aodn/ogcapi/server/core/service/das/DasServiceHeadersTest.java +++ b/server/src/test/java/au/org/aodn/ogcapi/server/core/service/das/DasServiceHeadersTest.java @@ -1,70 +1,97 @@ package au.org.aodn.ogcapi.server.core.service.das; import au.org.aodn.ogcapi.server.core.configuration.Config; +import au.org.aodn.ogcapi.server.core.http.RecordingSseConnector; +import au.org.aodn.ogcapi.server.core.util.DasSseFrames; +import com.fasterxml.jackson.core.type.TypeReference; import com.fasterxml.jackson.databind.ObjectMapper; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import org.springframework.http.HttpHeaders; +import org.springframework.http.HttpMethod; import org.springframework.http.MediaType; import org.springframework.test.web.client.MockRestServiceServer; import org.springframework.web.client.RestTemplate; +import org.springframework.web.reactive.function.client.WebClient; +import java.net.URI; import java.time.Duration; +import java.util.List; import java.util.Map; -import static org.springframework.test.web.client.match.MockRestRequestMatchers.content; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; import static org.springframework.test.web.client.match.MockRestRequestMatchers.header; import static org.springframework.test.web.client.match.MockRestRequestMatchers.headerDoesNotExist; -import static org.springframework.test.web.client.match.MockRestRequestMatchers.method; import static org.springframework.test.web.client.match.MockRestRequestMatchers.requestTo; import static org.springframework.test.web.client.response.MockRestResponseCreators.withSuccess; /** - * What DasService actually puts on the wire, driven through the real - * {@link Config#createDasRestTemplate} rather than a mocked RestTemplate — the mock-based - * {@link DasServiceTest} cannot see headers contributed by the client's interceptor or by the - * message converter that writes the body. + * What DasService actually puts on the wire, driven through the real clients Config builds. The + * mock-based DasServiceTest cannot see headers added by the client itself or by the body codec. */ public class DasServiceHeadersTest { private static final DasProperties PROPS = new DasProperties( "http://localhost:5000", null,"test-secret", "internal-secret", - Duration.ofSeconds(5), Duration.ofSeconds(30)); + Duration.ofSeconds(5), Duration.ofSeconds(30), Duration.ofMinutes(2)); + + private static final String RESULT_FRAME = """ + event: result + data: {"status":"completed","data":{"estimated_output_bytes":123}} + + """; private MockRestServiceServer server; + private RecordingSseConnector sseConnector; private DasService dasService; + /** + * The estimate goes out on the SSE client, so it is built the way Config builds it, with only + * the transport underneath swapped for a recording one. + */ + private static WebClient sseClientOn(DasProperties properties, RecordingSseConnector connector) { + return new Config().createDasSseWebClient(properties, new ObjectMapper()) + .mutate() + .clientConnector(connector) + .build(); + } + @BeforeEach public void setUp() { - RestTemplate template = new Config().createDasRestTemplate(PROPS); + Config config = new Config(); + RestTemplate template = config.createDasRestTemplate(PROPS); + sseConnector = new RecordingSseConnector(); + server = MockRestServiceServer.bindTo(template).build(); - dasService = new DasService(PROPS, template, new ObjectMapper()); + dasService = new DasService(PROPS, template, sseClientOn(PROPS, sseConnector), new ObjectMapper()); } @Test - public void testEstimateIsSentAsJsonNotXml() { - // Jackson's XML converter also claims Map bodies and is consulted before the JSON one, so - // without an explicit Content-Type this request goes out as application/xml and DAS breaks. - // The estimate is a streamed endpoint, so we accept text/event-stream and DAS replies with - // the payload wrapped in a terminal result frame. - String sseBody = """ - event: result - data: {"status":"completed","data":{"estimated_output_bytes":123}} - - """; - server.expect(requestTo("http://localhost:5000/api/v1/das/data/test-uuid/estimate_size")) - .andExpect(method(org.springframework.http.HttpMethod.POST)) - .andExpect(header("Content-Type", MediaType.APPLICATION_JSON_VALUE)) - .andExpect(header("Accept", MediaType.TEXT_EVENT_STREAM_VALUE)) - .andExpect(content().json("{\"uuid\":\"test-uuid\",\"output_format\":\"netcdf\"}")) - // the per-call headers above must not displace the ones the client attaches - .andExpect(header("X-API-KEY", "test-secret")) - .andExpect(header("x-internal-das-header-secret", "internal-secret")) - .andRespond(withSuccess(sseBody, MediaType.TEXT_EVENT_STREAM)); + public void testEstimateIsSentAsJsonNotXml() throws Exception { + // Jackson's XML encoder also claims Map bodies, so without an explicit Content-Type this + // could go out as application/xml and DAS would break. It is a streamed endpoint, so we + // accept text/event-stream and DAS replies with the payload in a terminal result frame. + sseConnector.respondWith(List.of(RESULT_FRAME)); dasService.estimateCloudOptimisedDownloadSize( - "test-uuid", Map.of("uuid", "test-uuid", "output_format", "netcdf")); - - server.verify(); + "test-uuid", Map.of("uuid", "test-uuid", "output_format", "netcdf"), + DasSseFrames.FrameCallback.IGNORE); + + assertEquals(HttpMethod.POST, sseConnector.method()); + assertEquals(URI.create("http://localhost:5000/api/v1/das/data/test-uuid/estimate_size"), + sseConnector.uri()); + + HttpHeaders headers = sseConnector.headers(); + assertEquals(MediaType.APPLICATION_JSON_VALUE, headers.getFirst(HttpHeaders.CONTENT_TYPE)); + assertEquals(MediaType.TEXT_EVENT_STREAM_VALUE, headers.getFirst(HttpHeaders.ACCEPT)); + // the per-call headers above must not displace the ones the client attaches + assertEquals("test-secret", headers.getFirst("X-API-KEY")); + assertEquals("internal-secret", headers.getFirst("x-internal-das-header-secret")); + + assertEquals(Map.of("uuid", "test-uuid", "output_format", "netcdf"), + new ObjectMapper().readValue(sseConnector.body(), new TypeReference>() { + })); } @Test @@ -83,16 +110,25 @@ public void testFeatureCollectionCarriesCredentials() { public void testInternalSecretIsOmittedWhenNotConfigured() { DasProperties noInternal = new DasProperties( "http://localhost:5000", null,"test-secret", null, - Duration.ofSeconds(5), Duration.ofSeconds(30)); - RestTemplate noInternalTemplate = new Config().createDasRestTemplate(noInternal); + Duration.ofSeconds(5), Duration.ofSeconds(30), Duration.ofMinutes(2)); + Config config = new Config(); + RestTemplate noInternalTemplate = config.createDasRestTemplate(noInternal); MockRestServiceServer noInternalServer = MockRestServiceServer.bindTo(noInternalTemplate).build(); noInternalServer.expect(header("X-API-KEY", "test-secret")) .andExpect(headerDoesNotExist("x-internal-das-header-secret")) .andRespond(withSuccess("{}", MediaType.APPLICATION_JSON)); - new DasService(noInternal, noInternalTemplate, new ObjectMapper()).getWaveBuoysLatestAvailableDate(); + RecordingSseConnector connector = new RecordingSseConnector().respondWith(List.of(RESULT_FRAME)); + DasService service = new DasService( + noInternal, noInternalTemplate, sseClientOn(noInternal, connector), new ObjectMapper()); + service.getWaveBuoysLatestAvailableDate(); noInternalServer.verify(); + + // The streamed client is built from the same properties, so it must leave it off too. + service.estimateCloudOptimisedDownloadSize( + "test-uuid", Map.of("uuid", "test-uuid"), DasSseFrames.FrameCallback.IGNORE); + assertNull(connector.headers().getFirst("x-internal-das-header-secret")); } } diff --git a/server/src/test/java/au/org/aodn/ogcapi/server/core/service/das/DasServiceTest.java b/server/src/test/java/au/org/aodn/ogcapi/server/core/service/das/DasServiceTest.java index f0f21077..4ed8de45 100644 --- a/server/src/test/java/au/org/aodn/ogcapi/server/core/service/das/DasServiceTest.java +++ b/server/src/test/java/au/org/aodn/ogcapi/server/core/service/das/DasServiceTest.java @@ -4,19 +4,14 @@ import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.mockito.ArgumentCaptor; -import org.springframework.http.HttpEntity; -import org.springframework.http.HttpHeaders; -import org.springframework.http.HttpStatus; -import org.springframework.http.MediaType; import org.springframework.http.ResponseEntity; -import org.springframework.web.client.HttpClientErrorException; import org.springframework.web.client.RestTemplate; +import org.springframework.web.reactive.function.client.WebClient; import java.time.Duration; import java.util.Map; import static org.junit.jupiter.api.Assertions.*; -import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyMap; import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.ArgumentMatchers.eq; @@ -25,11 +20,10 @@ import static org.mockito.Mockito.when; /** - * Unit tests for DasService URL building. Verifies that null date params are omitted (so they are - * never passed to URI template expansion) and that buoy/mooring identifiers are sent as path - * variables for single, correct encoding. Also covers the cloud-optimised size-estimate call - * (request body shape, SSE unwrapping and error propagation). The API key is attached by the - * RestTemplate bean, so it is covered by ConfigTest rather than here. + * Unit tests for DasService URL building: null date params are omitted so they never reach URI + * template expansion, and buoy/mooring ids go as path variables so they are encoded once. The API + * key is attached by the RestTemplate bean, so ConfigTest covers it, and the streamed size + * estimate has its own DasServiceEstimateStreamTest. */ public class DasServiceTest { @@ -44,10 +38,10 @@ public void setUp() { DasProperties config = new DasProperties( HOST, null,"test-secret", "", - Duration.ofSeconds(5), Duration.ofSeconds(30) + Duration.ofSeconds(5), Duration.ofSeconds(30), Duration.ofMinutes(20) ); - dasService = new DasService(config, httpClient, new ObjectMapper()); + dasService = new DasService(config, httpClient, mock(WebClient.class), new ObjectMapper()); when(httpClient.getForEntity(anyString(), eq(byte[].class), anyMap())) .thenReturn(ResponseEntity.ok("ok".getBytes())); @@ -124,100 +118,6 @@ public void testLatestAvailableDateUsesNoUriVariables() { assertEquals(HOST + "/api/v1/das/data/feature-collection/wave-buoy/latest", urlCaptor.getValue()); } - /** - * Mocks a successful SSE estimate response, invokes the estimate, asserts the payload is - * unwrapped from the terminal {@code result} frame, and returns the captured request entity so - * callers can assert on the forwarded headers / body. - */ - @SuppressWarnings("unchecked") - private HttpEntity> callEstimateAndCaptureEntity(String uuid, Map parameters) { - - // DAS streams the estimate: heartbeats while it computes, then the dict nested - // under the terminal result event. - String sseBody = """ - event: processing - data: {"status":"processing","message":"Processing your request..."} - - event: result - data: {"status":"completed","message":"Done","data":{"estimated_output_bytes":123}} - - """; - when(httpClient.postForObject(anyString(), any(), eq(String.class), anyMap())) - .thenReturn(sseBody); - - String result = dasService.estimateCloudOptimisedDownloadSize(uuid, parameters); - assertEquals("{\"estimated_output_bytes\":123}", result, "The estimate dict should be unwrapped from the stream"); - - ArgumentCaptor urlCaptor = ArgumentCaptor.forClass(String.class); - ArgumentCaptor entityCaptor = ArgumentCaptor.forClass(HttpEntity.class); - ArgumentCaptor> uriVarsCaptor = ArgumentCaptor.forClass(Map.class); - verify(httpClient).postForObject(urlCaptor.capture(), entityCaptor.capture(), eq(String.class), uriVarsCaptor.capture()); - - assertEquals(HOST + "/api/v1/das/data/{uuid}/estimate_size", urlCaptor.getValue()); - assertEquals(uuid, uriVarsCaptor.getValue().get("uuid")); - - return (HttpEntity>) entityCaptor.getValue(); - } - - @Test - public void testEstimatePostsBatchStyleParametersUnchanged() { - // The estimate forwards the same batch-style subset parameter map to DAS unchanged. - Map parameters = Map.of( - "uuid", "test-uuid", - "key", "a.zarr,b.zarr", - "start_date", "2023-01-01", - "end_date", "2023-01-31", - "multi_polygon", "non-specified", - "output_format", "netcdf"); - - HttpEntity> entity = callEstimateAndCaptureEntity("test-uuid", parameters); - - assertEquals(parameters, entity.getBody(), - "The batch-style parameter map must be forwarded to DAS unchanged"); - } - - @Test - public void testEstimateAcceptsEventStream() { - HttpEntity> entity = callEstimateAndCaptureEntity( - "test-uuid", Map.of("uuid", "test-uuid", "output_format", "netcdf")); - - assertEquals(MediaType.TEXT_EVENT_STREAM_VALUE, entity.getHeaders().getFirst(HttpHeaders.ACCEPT)); - } - - @Test - public void testEstimateErrorEventThrows() { - // A failure raised after the stream opened comes back on an HTTP 200, so it is - // only visible in the frames. It must still surface as a thrown exception, - // otherwise the SSE layer would report a failed estimate as a successful one. - String sseBody = """ - event: processing - data: {"status":"processing","message":"Processing your request..."} - - event: error - data: {"status":"error","message":"404: No matching keys found for uuid=bad-uuid"} - - """; - when(httpClient.postForObject(anyString(), any(), eq(String.class), anyMap())) - .thenReturn(sseBody); - - RuntimeException e = assertThrows(RuntimeException.class, () -> - dasService.estimateCloudOptimisedDownloadSize( - "bad-uuid", Map.of("uuid", "bad-uuid", "key", "missing.zarr", "output_format", "netcdf"))); - - assertEquals("404: No matching keys found for uuid=bad-uuid", e.getMessage(), - "DAS's reason is forwarded verbatim for the SSE layer to report"); - } - - @Test - public void testEstimateNon2xxPropagates() { - when(httpClient.postForObject(anyString(), any(), eq(String.class), anyMap())) - .thenThrow(HttpClientErrorException.create(HttpStatus.NOT_FOUND, "Not Found", HttpHeaders.EMPTY, null, null)); - - assertThrows(HttpClientErrorException.class, () -> - dasService.estimateCloudOptimisedDownloadSize( - "bad-uuid", Map.of("uuid", "bad-uuid", "key", "missing.zarr", "output_format", "netcdf"))); - } - private record CapturedRequest(String url, Map params) { } } diff --git a/server/src/test/java/au/org/aodn/ogcapi/server/core/service/das/DasTilerServiceTest.java b/server/src/test/java/au/org/aodn/ogcapi/server/core/service/das/DasTilerServiceTest.java index 6eef32ad..75ec366e 100644 --- a/server/src/test/java/au/org/aodn/ogcapi/server/core/service/das/DasTilerServiceTest.java +++ b/server/src/test/java/au/org/aodn/ogcapi/server/core/service/das/DasTilerServiceTest.java @@ -49,7 +49,7 @@ public void setUp() { DasProperties config = new DasProperties( HOST, null,"test-secret", "internal-secret", - Duration.ofSeconds(5), Duration.ofSeconds(30) + Duration.ofSeconds(5), Duration.ofSeconds(30), Duration.ofMinutes(20) ); service = new DasTilerService(config, httpClient, new ObjectMapper()); diff --git a/server/src/test/java/au/org/aodn/ogcapi/server/core/exception/wfs/WfsErrorHandlerTest.java b/server/src/test/java/au/org/aodn/ogcapi/server/core/service/sse/SseErrorHandlerTest.java similarity index 88% rename from server/src/test/java/au/org/aodn/ogcapi/server/core/exception/wfs/WfsErrorHandlerTest.java rename to server/src/test/java/au/org/aodn/ogcapi/server/core/service/sse/SseErrorHandlerTest.java index 27d64dbc..0da54a6f 100644 --- a/server/src/test/java/au/org/aodn/ogcapi/server/core/exception/wfs/WfsErrorHandlerTest.java +++ b/server/src/test/java/au/org/aodn/ogcapi/server/core/service/sse/SseErrorHandlerTest.java @@ -1,4 +1,4 @@ -package au.org.aodn.ogcapi.server.core.exception.wfs; +package au.org.aodn.ogcapi.server.core.service.sse; import au.org.aodn.ogcapi.server.core.exception.GeoserverFieldsNotFoundException; import au.org.aodn.ogcapi.server.core.exception.UnauthorizedServerException; @@ -18,22 +18,22 @@ import static org.junit.jupiter.api.Assertions.*; import static org.mockito.Mockito.*; -public class WfsErrorHandlerTest { +public class SseErrorHandlerTest { protected TestLogAppender logAppender; @BeforeEach void attachLogAppender() { - logAppender = TestLogAppender.attachTo(WfsErrorHandler.class); + logAppender = TestLogAppender.attachTo(SseErrorHandler.class); } @AfterEach void detachLogAppender() { - logAppender.detachFrom(WfsErrorHandler.class); + logAppender.detachFrom(SseErrorHandler.class); } /** - * An upstream WFS server error status must be logged at ERROR level (so New + * An upstream server error status must be logged at ERROR level (so New * Relic can pick it up), sent to the client as an SSE error event, and must * not be mistaken for a client disconnect. */ @@ -41,7 +41,7 @@ void detachLogAppender() { void verifyUpstreamServerErrorLoggedAtErrorLevel() throws IOException { SseEmitter emitter = mock(SseEmitter.class); - WfsErrorHandler.handleError( + SseErrorHandler.handleError( new HttpServerErrorException(HttpStatus.INTERNAL_SERVER_ERROR), "uuid-123", emitter, null); @@ -61,7 +61,7 @@ void verifyUpstreamServerErrorLoggedAtErrorLevel() throws IOException { void verifyUnknownErrorLoggedAtErrorLevel() throws IOException { SseEmitter emitter = mock(SseEmitter.class); - WfsErrorHandler.handleError(new RuntimeException("boom"), "uuid-123", emitter, null); + SseErrorHandler.handleError(new RuntimeException("boom"), "uuid-123", emitter, null); assertEquals(1, logAppender.eventsAtLevel(Level.ERROR).size()); verify(emitter, times(1)).send(any(SseEmitter.SseEventBuilder.class)); @@ -69,7 +69,7 @@ void verifyUnknownErrorLoggedAtErrorLevel() throws IOException { } /** - * A WFS server that is not authorized must be logged at WARN with the + * An upstream server that is not authorized must be logged at WARN with the * exception attached (its message names the server) and reported to the * client, without raising an ERROR alert. */ @@ -77,7 +77,7 @@ void verifyUnknownErrorLoggedAtErrorLevel() throws IOException { void verifyUnauthorizedServerLoggedAtWarnLevel() throws IOException { SseEmitter emitter = mock(SseEmitter.class); - WfsErrorHandler.handleError( + SseErrorHandler.handleError( new UnauthorizedServerException("Server http://not-allowed/wfs is not authorized"), "uuid-123", emitter, null); @@ -100,7 +100,7 @@ void verifyUnauthorizedServerLoggedAtWarnLevel() throws IOException { void verifyDownloadableFieldsNotFoundLoggedAtWarnLevel() throws IOException { SseEmitter emitter = mock(SseEmitter.class); - WfsErrorHandler.handleError( + SseErrorHandler.handleError( new GeoserverFieldsNotFoundException("No downloadable fields found for all url"), "uuid-123", emitter, null); @@ -122,7 +122,7 @@ void verifyDownloadableFieldsNotFoundLoggedAtWarnLevel() throws IOException { void verifyClientDisconnectStaysAtWarnLevel() { SseEmitter emitter = mock(SseEmitter.class); - WfsErrorHandler.handleError(new IOException("Broken pipe"), "uuid-123", emitter, null); + SseErrorHandler.handleError(new IOException("Broken pipe"), "uuid-123", emitter, null); assertTrue(logAppender.eventsAtLevel(Level.ERROR).isEmpty()); assertEquals(1, logAppender.eventsAtLevel(Level.WARN).size()); diff --git a/server/src/test/java/au/org/aodn/ogcapi/server/core/util/DasSseFramesTest.java b/server/src/test/java/au/org/aodn/ogcapi/server/core/util/DasSseFramesTest.java new file mode 100644 index 00000000..c5cf92ca --- /dev/null +++ b/server/src/test/java/au/org/aodn/ogcapi/server/core/util/DasSseFramesTest.java @@ -0,0 +1,102 @@ +package au.org.aodn.ogcapi.server.core.util; + +import com.fasterxml.jackson.databind.ObjectMapper; +import org.junit.jupiter.api.Test; +import org.springframework.http.codec.ServerSentEvent; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * What DAS means by a frame. These mirror what its sse_it decorator emits, including the error + * frames that arrive on an HTTP 200 instead of an error status. Splitting frames out of the + * stream is the HTTP client's job and is covered by DasServiceEstimateStreamTest. + */ +public class DasSseFramesTest { + + private final ObjectMapper objectMapper = new ObjectMapper(); + + private static ServerSentEvent frame(String event, String data) { + return ServerSentEvent.builder().event(event).data(data).build(); + } + + private String read(String event, String data) { + return DasSseFrames.readTerminalFrame(objectMapper, frame(event, data)); + } + + @Test + public void testResultFrameReturnsNestedData() { + assertEquals("{\"uuid\":\"abc\",\"estimated_output_bytes\":123}", + read("result", "{\"status\":\"completed\",\"message\":\"Done\",\"data\":{\"uuid\":\"abc\",\"estimated_output_bytes\":123}}"), + "The estimate dict nested under the result event's data field should be returned"); + } + + @Test + public void testLargeByteCountSurvivesUnchanged() { + // The estimate is a Python int with no width limit, so it must not be routed + // through a lossy numeric type on the way out. + String payload = read("result", "{\"status\":\"completed\",\"data\":{\"estimated_output_bytes\":9007199254740993}}"); + + assertTrue(payload.contains("9007199254740993"), "A byte count beyond 2^53 must not lose precision"); + } + + @Test + public void testHeartbeatIsSkipped() { + // A slow estimate heartbeats until the work finishes; null tells the caller to read on. + assertNull(read("processing", "{\"status\":\"processing\",\"message\":\"Processing your request...\"}")); + } + + @Test + public void testFrameWithNoEventNameIsSkipped() { + assertNull(read(null, "{\"status\":\"processing\"}")); + } + + @Test + public void testErrorFrameThrowsWithDasMessageVerbatim() { + // What a "no matching keys" failure looks like now the route raises inside the + // stream: HTTP 200, and Starlette's HTTPException.__str__ supplies the "404: ". + RuntimeException e = assertThrows(RuntimeException.class, () -> + read("error", "{\"status\":\"error\",\"message\":\"404: No matching keys found for uuid=abc, keys=['missing.zarr']\"}")); + + assertEquals("404: No matching keys found for uuid=abc, keys=['missing.zarr']", e.getMessage(), + "DAS's reason must be rethrown unmodified; callers add their own context"); + } + + @Test + public void testErrorFrameWithoutAMessageStillThrows() { + RuntimeException e = assertThrows(RuntimeException.class, () -> read("error", "{\"status\":\"error\"}")); + + assertTrue(e.getMessage().contains("no message"), "Got: " + e.getMessage()); + } + + @Test + public void testResultFrameWithoutDataThrows() { + RuntimeException e = assertThrows(RuntimeException.class, () -> + read("result", "{\"status\":\"completed\",\"message\":\"Done\"}")); + + assertTrue(e.getMessage().contains("carried no data"), "Got: " + e.getMessage()); + } + + @Test + public void testResultFrameWithNullDataThrows() { + RuntimeException e = assertThrows(RuntimeException.class, () -> + read("result", "{\"status\":\"completed\",\"data\":null}")); + + assertTrue(e.getMessage().contains("carried no data"), "Got: " + e.getMessage()); + } + + @Test + public void testUnreadableTerminalFrameThrows() { + RuntimeException e = assertThrows(RuntimeException.class, () -> read("result", "{\"status\":\"comp")); + + assertTrue(e.getMessage().contains("Unreadable"), "Got: " + e.getMessage()); + } + + @Test + public void testTerminalFrameWithNoDataAtAllThrows() { + // A result event with no data line reads as an empty document, not as a heartbeat. + assertThrows(RuntimeException.class, () -> read("result", null)); + } +} diff --git a/server/src/test/java/au/org/aodn/ogcapi/server/core/util/SseResponseParserTest.java b/server/src/test/java/au/org/aodn/ogcapi/server/core/util/SseResponseParserTest.java deleted file mode 100644 index 726a65b1..00000000 --- a/server/src/test/java/au/org/aodn/ogcapi/server/core/util/SseResponseParserTest.java +++ /dev/null @@ -1,167 +0,0 @@ -package au.org.aodn.ogcapi.server.core.util; - -import com.fasterxml.jackson.databind.ObjectMapper; -import org.junit.jupiter.api.Test; - -import static org.junit.jupiter.api.Assertions.*; - -/** - * Unit tests for the data-access-service SSE body parser. The frames used here mirror - * what DAS's {@code sse_it} decorator emits ({@code format_sse} writes - * {@code "event: \ndata: \n\n"}), including the error frames that now - * arrive on an HTTP 200 in place of the status codes the endpoint used to return. - */ -public class SseResponseParserTest { - - private final ObjectMapper objectMapper = new ObjectMapper(); - - private String parse(String body) { - return SseResponseParser.extractResultData(objectMapper, body); - } - - @Test - public void testResultFrameReturnsNestedData() { - String body = """ - event: processing - data: {"status":"processing","message":"Processing your request..."} - - event: result - data: {"status":"completed","message":"Done","data":{"uuid":"abc","estimated_output_bytes":123}} - - """; - - assertEquals("{\"uuid\":\"abc\",\"estimated_output_bytes\":123}", parse(body), - "The estimate dict nested under the result event's data field should be returned"); - } - - @Test - public void testLargeByteCountSurvivesUnchanged() { - // The estimate is a Python int with no width limit, so it must not be routed - // through a lossy numeric type on the way out. - String body = """ - event: result - data: {"status":"completed","data":{"estimated_output_bytes":9007199254740993}} - - """; - - assertTrue(parse(body).contains("9007199254740993"), "A byte count beyond 2^53 must not lose precision"); - } - - @Test - public void testHeartbeatsBeforeResultAreSkipped() { - // A slow estimate heartbeats every 30s until the work finishes. - String body = """ - event: processing - data: {"status":"processing","message":"Processing your request..."} - - event: processing - data: {"status":"processing","message":"Still processing..."} - - event: processing - data: {"status":"processing","message":"Still processing..."} - - event: result - data: {"status":"completed","message":"Done","data":{"estimated_output_bytes":1}} - - """; - - assertEquals("{\"estimated_output_bytes\":1}", parse(body)); - } - - @Test - public void testErrorFrameThrowsWithDasMessageVerbatim() { - // What a "no matching keys" failure looks like now the route raises inside the - // stream: HTTP 200, and Starlette's HTTPException.__str__ supplies the "404: ". - String body = """ - event: processing - data: {"status":"processing","message":"Processing your request..."} - - event: error - data: {"status":"error","message":"404: No matching keys found for uuid=abc, keys=['missing.zarr']"} - - """; - - RuntimeException e = assertThrows(RuntimeException.class, () -> parse(body)); - assertEquals("404: No matching keys found for uuid=abc, keys=['missing.zarr']", e.getMessage(), - "DAS's reason must be rethrown unmodified; callers add their own context"); - } - - @Test - public void testHeartbeatOnlyStreamThrows() { - // The connection dropped before the estimate finished. - String body = """ - event: processing - data: {"status":"processing","message":"Processing your request..."} - - event: processing - data: {"status":"processing","message":"Still processing..."} - - """; - - RuntimeException e = assertThrows(RuntimeException.class, () -> parse(body)); - assertTrue(e.getMessage().contains("without a result or error event"), "Got: " + e.getMessage()); - } - - @Test - public void testTruncatedStreamThrows() { - String body = "event: processing\ndata: {\"status\":\"proce"; - - assertThrows(RuntimeException.class, () -> parse(body)); - } - - @Test - public void testResultFrameWithoutDataThrows() { - String body = """ - event: result - data: {"status":"completed","message":"Done"} - - """; - - RuntimeException e = assertThrows(RuntimeException.class, () -> parse(body)); - assertTrue(e.getMessage().contains("carried no data"), "Got: " + e.getMessage()); - } - - @Test - public void testTerminalFrameNotFollowedByBlankLineIsStillRead() { - String body = "event: result\ndata: {\"status\":\"completed\",\"data\":{\"estimated_output_bytes\":7}}"; - - assertEquals("{\"estimated_output_bytes\":7}", parse(body)); - } - - @Test - public void testCarriageReturnLineEndingsAreHandled() { - String body = "event: processing\r\ndata: {\"status\":\"processing\"}\r\n" + - "\r\n" + - "event: result\r\ndata: {\"status\":\"completed\",\"data\":{\"estimated_output_bytes\":7}}\r\n\r\n"; - - assertEquals("{\"estimated_output_bytes\":7}", parse(body)); - } - - @Test - public void testCommentLinesAreIgnored() { - String body = """ - : this is a keep-alive comment - - event: result - data: {"status":"completed","data":{"estimated_output_bytes":7}} - - """; - - assertEquals("{\"estimated_output_bytes\":7}", parse(body)); - } - - @Test - public void testNonSseBodyThrows() { - // The endpoint only speaks SSE now; a bare JSON body has no terminal frame. - String body = "{\"uuid\":\"abc\",\"estimated_output_bytes\":123}"; - - RuntimeException e = assertThrows(RuntimeException.class, () -> parse(body)); - assertTrue(e.getMessage().contains("without a result or error event"), "Got: " + e.getMessage()); - } - - @Test - public void testEmptyBodyThrows() { - assertThrows(RuntimeException.class, () -> parse("")); - assertThrows(RuntimeException.class, () -> parse(null)); - } -} diff --git a/server/src/test/java/au/org/aodn/ogcapi/server/processes/RestApiSseTest.java b/server/src/test/java/au/org/aodn/ogcapi/server/processes/RestApiSseTest.java index c22bcb44..babbf7ad 100644 --- a/server/src/test/java/au/org/aodn/ogcapi/server/processes/RestApiSseTest.java +++ b/server/src/test/java/au/org/aodn/ogcapi/server/processes/RestApiSseTest.java @@ -1,10 +1,14 @@ package au.org.aodn.ogcapi.server.processes; +import au.org.aodn.ogcapi.server.core.exception.SseClientGoneException; +import au.org.aodn.ogcapi.server.core.service.sse.SseErrorHandler; import au.org.aodn.ogcapi.server.core.model.enumeration.DatasetDownloadEnums; import au.org.aodn.ogcapi.server.core.model.enumeration.ProcessIdEnum; import au.org.aodn.ogcapi.server.core.service.das.DasService; import au.org.aodn.ogcapi.server.core.service.geoserver.wfs.DownloadWfsDataService; +import au.org.aodn.ogcapi.server.core.util.TestLogAppender; import com.fasterxml.jackson.databind.ObjectMapper; +import org.apache.logging.log4j.Level; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; @@ -19,9 +23,12 @@ import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; import software.amazon.awssdk.services.batch.BatchClient; +import java.io.IOException; +import java.io.UncheckedIOException; import java.math.BigInteger; import java.util.HashMap; import java.util.Map; +import java.util.stream.Collectors; import static org.junit.jupiter.api.Assertions.*; import static org.mockito.ArgumentMatchers.*; @@ -74,11 +81,9 @@ private MockHttpServletResponse postSse(String processId, Map in } /** - * The SSE work runs on a separate thread, so poll the mock response until the - * expected marker shows up (or time out and let the caller's assert fail with - * the content collected so far). The emitter writes an event's name and data - * lines separately, so also wait for the blank line that terminates the - * marker's event — otherwise callers could assert on a half-written payload. + * The SSE work runs on another thread, so poll the mock response until the marker appears, or + * time out and let the caller's assert fail with what was collected. Wait for the blank line + * ending the event too, or callers could assert on a half-written payload. */ private String awaitContent(MockHttpServletResponse response, String expectedMarker) throws Exception { long deadline = System.currentTimeMillis() + 5000; @@ -104,7 +109,7 @@ private static void assertEventOrder(String content, String earlierEvent, String @Test public void testEstimateCODownloadForwardsBatchStyleParameters() throws Exception { String dasJson = "{\"estimated_output_bytes\":12345}"; - when(dasService.estimateCloudOptimisedDownloadSize(any(), anyMap())) + when(dasService.estimateCloudOptimisedDownloadSize(any(), anyMap(), any())) .thenReturn(dasJson); Map inputs = new HashMap<>(); @@ -123,7 +128,7 @@ public void testEstimateCODownloadForwardsBatchStyleParameters() throws Exceptio @SuppressWarnings("unchecked") ArgumentCaptor> paramsCaptor = ArgumentCaptor.forClass(Map.class); - verify(dasService).estimateCloudOptimisedDownloadSize(eq("test-uuid"), paramsCaptor.capture()); + verify(dasService).estimateCloudOptimisedDownloadSize(eq("test-uuid"), paramsCaptor.capture(), any()); Map params = paramsCaptor.getValue(); // key is forwarded raw (CSV, untrimmed) - DAS splits it, matching the batch download. @@ -136,7 +141,7 @@ public void testEstimateCODownloadForwardsBatchStyleParameters() throws Exceptio @Test public void testEstimateCODownloadForwardsWildcardKeyRaw() throws Exception { - when(dasService.estimateCloudOptimisedDownloadSize(any(), anyMap())) + when(dasService.estimateCloudOptimisedDownloadSize(any(), anyMap(), any())) .thenReturn("{}"); Map inputs = new HashMap<>(); @@ -150,14 +155,14 @@ public void testEstimateCODownloadForwardsWildcardKeyRaw() throws Exception { @SuppressWarnings("unchecked") ArgumentCaptor> paramsCaptor = ArgumentCaptor.forClass(Map.class); - verify(dasService).estimateCloudOptimisedDownloadSize(eq("test-uuid"), paramsCaptor.capture()); + verify(dasService).estimateCloudOptimisedDownloadSize(eq("test-uuid"), paramsCaptor.capture(), any()); assertEquals("*", paramsCaptor.getValue().get(DatasetDownloadEnums.Parameter.KEY.getValue()), "wildcard key is forwarded raw; DAS expands it to all keys"); } @Test public void testEstimateCODownloadDasFailureEmitsEstimateFailed() throws Exception { - when(dasService.estimateCloudOptimisedDownloadSize(any(), anyMap())) + when(dasService.estimateCloudOptimisedDownloadSize(any(), anyMap(), any())) .thenThrow(new RuntimeException("das returned 404")); Map inputs = new HashMap<>(); @@ -172,6 +177,52 @@ public void testEstimateCODownloadDasFailureEmitsEstimateFailed() throws Excepti assertTrue(content.contains("das returned 404"), "Failure reason should be forwarded in: " + content); } + @Test + public void testEstimateCODownloadClientDisconnectIsNotReportedAsAFailedEstimate() throws Exception { + // A disconnect is what aborts the DAS call, and the estimate has an unchecked signature, + // so the IOException that unwound it arrives nested. No client is left to tell, and + // reporting estimate-failed here would log a failure that never happened. + when(dasService.estimateCloudOptimisedDownloadSize(any(), anyMap(), any())) + .thenThrow(new UncheckedIOException( + new SseClientGoneException("test-uuid", new IOException("Broken pipe")))); + + TestLogAppender logs = TestLogAppender.attachTo(SseErrorHandler.class); + try { + Map inputs = new HashMap<>(); + inputs.put(DatasetDownloadEnums.Parameter.UUID.getValue(), "test-uuid"); + inputs.put(DatasetDownloadEnums.Parameter.MULTI_POLYGON.getValue(), "non-specified"); + inputs.put(DatasetDownloadEnums.Parameter.OUTPUT_FORMAT.getValue(), "netcdf"); + + MockHttpServletResponse response = postSse(ProcessIdEnum.DOWNLOAD_CO_ESTIMATE.getValue(), inputs); + + String disconnectLog = awaitLogContaining(logs, "Client disconnected for UUID: test-uuid"); + assertTrue(disconnectLog.contains("Client disconnected"), + "The disconnect should be logged as a disconnect, got: " + disconnectLog); + assertFalse(response.getContentAsString().contains("event:estimate-failed"), + "A departed client must not be told its estimate failed: " + response.getContentAsString()); + } finally { + logs.detachFrom(SseErrorHandler.class); + } + } + + /** + * The stream's work runs on another thread, so wait for the expected line rather than + * assuming it has been logged by the time the request returns. + */ + private String awaitLogContaining(TestLogAppender logs, String expected) throws Exception { + long deadline = System.currentTimeMillis() + 5000; + while (System.currentTimeMillis() < deadline) { + String messages = logs.eventsAtLevel(Level.WARN).stream() + .map(event -> event.getMessage().getFormattedMessage()) + .collect(Collectors.joining("\n")); + if (messages.contains(expected)) { + return messages; + } + Thread.sleep(50); + } + return ""; + } + @Test public void testEstimateCODownloadMissingUuidEmitsError() throws Exception { Map inputs = new HashMap<>();