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 22c58026..2477fff7 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 @@ -4,6 +4,7 @@ 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.DasSseFrames; +import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.core.ParameterizedTypeReference; @@ -30,6 +31,8 @@ public class DasService implements ApplicationInfo { new ParameterizedTypeReference<>() { }; + private static final int MAX_UPSTREAM_DETAIL_CHARS = 200; + protected final DasProperties dasProperties; protected final RestTemplate httpClient; protected final WebClient sseHttpClient; @@ -156,14 +159,40 @@ public String estimateCloudOptimisedDownloadSize(String uuid, } /** - * 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. + * Describe a failure that arrived before the stream opened. */ - private static String describe(HttpStatusCode status, String body) { + private 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; + String detail = reasonFrom(body); + return detail.isEmpty() ? failure : failure + ": " + detail; + } + + /** + * Pull the reason out of an error body. + */ + private String reasonFrom(String body) { + String flattened = body.replaceAll("\\s+", " ").trim(); + if (flattened.isEmpty()) { + return ""; + } + + try { + JsonNode detail = objectMapper.readTree(flattened).get("detail"); + if (detail != null && !detail.isNull()) { + return truncate(detail.isTextual() ? detail.asText() : detail.toString()); + } + } catch (Exception ignored) { + // Not JSON, so not DAS speaking. Fall through and quote what little is useful. + } + + return flattened.startsWith("<") ? "" : truncate(flattened); + } + + private static String truncate(String detail) { + return detail.length() <= MAX_UPSTREAM_DETAIL_CHARS ? + detail : + detail.substring(0, MAX_UPSTREAM_DETAIL_CHARS) + "..."; } public ResponseEntity getDatasetMetadata(String datasetId) { 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 8a4cd5ab..6d94dab7 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 @@ -11,6 +11,7 @@ import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicReference; import java.util.function.Supplier; @@ -24,6 +25,10 @@ @Slf4j public class SseSession { + // What a probe writes. Only ever read by whatever is proxying this stream, which cares that + // bytes moved and not what they say, so it is short. + private static final String PROBE_COMMENT = "probe"; + private final String contextId; @Getter @@ -32,6 +37,11 @@ public class SseSession { private final AtomicReference> keepAliveTaskRef = new AtomicReference<>(); private final AtomicReference keepAliveExecutorRef = new AtomicReference<>(); + // When an event last reached the client, so the keep-alive can tell a quiet stream from a + // busy one. Probes are writes but not events and deliberately do not count, see probeClient. + // Starts at creation: nothing has been sent yet, but nothing is overdue either. + private final AtomicLong lastEventSentAt = new AtomicLong(System.currentTimeMillis()); + public SseSession(String contextId, SseEmitter emitter) { this.contextId = contextId; this.emitter = emitter; @@ -42,32 +52,38 @@ public SseSession(String contextId, SseEmitter emitter) { */ public void send(SseEventName eventName, Object data) throws IOException { emitter.send(SseEmitter.event().name(eventName.getValue()).data(data)); + lastEventSentAt.set(System.currentTimeMillis()); } /** - * 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. + * Write to the client and report a dead one as SseClientGoneException. + * 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. + * It writes an SSE comment, not an event: still a real write, but the browser ignores it and + * the keep-alive ticker does not count it as activity. */ - public void probeClient(Object data) throws SseClientGoneException { + public void probeClient() throws SseClientGoneException { try { - send(SseEventName.KEEP_ALIVE, data); + emitter.send(SseEmitter.event().comment(PROBE_COMMENT)); } 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 - * reflect changing state (e.g. whether an upstream server has responded yet). + * Keep the client's connection busy with a keep-alive event every intervalSeconds, skipping + * a tick when an event was sent within the last half interval so work that reports its own + * progress is not doubled up on. payloadSupplier is called each tick, so the payload can + * reflect current state. */ public void startKeepAlive(long intervalSeconds, Supplier payloadSupplier) { + long quietEnoughMillis = intervalSeconds * 500L; ScheduledExecutorService keepAliveExecutor = Executors.newSingleThreadScheduledExecutor(); ScheduledFuture keepAliveTask = keepAliveExecutor.scheduleAtFixedRate(() -> { try { + if (System.currentTimeMillis() - lastEventSentAt.get() < quietEnoughMillis) { + return; + } send(SseEventName.KEEP_ALIVE, payloadSupplier.get()); } catch (Exception e) { // This only ends the ticker and the emitter: a disconnect noticed here cannot 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 6fd7c363..c94e0ed3 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 @@ -37,6 +37,15 @@ public class RestServices { private final String batchJobDefinition; private final String batchJobQueue; + // How often the cloud-optimised estimate reassures its client while data-access-service + // works. It has to stay well under the shortest idle timeout in front of this service - + // CloudFront allows 30s between packets - so a DAS that fails without ever answering cannot + // leave the connection quiet long enough to be dropped before the failure is reported. + // Nothing else emits keep-alive on this stream, so this is the rate the client sees, and it + // does not matter that it happens to match the DAS heartbeat. Lowered in tests to keep them + // fast. + private long estimateKeepAliveSeconds = 5; + @Autowired private DownloadWfsDataService downloadWfsDataService; @@ -332,16 +341,19 @@ public SseEmitter estimateCloudOptimisedDownloadWithSse(String uuid, "timestamp", System.currentTimeMillis() )); - // 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. + // STEP 2: Keep the connection alive while DAS works, two ways. + // 1. This ticker, one keep-alive per interval, so the client keeps receiving bytes. + // 2. probeClient on each DAS heartbeat below, so a client that has gone stops the + // DAS call instead of leaving it running. + session.startKeepAlive(estimateKeepAliveSeconds, () -> Map.of( + "message", "Estimating download size...", + "timestamp", System.currentTimeMillis() + )); + + // STEP 3: Call the data-access-service estimate endpoint and forward the result try { - String estimateJson = dasService.estimateCloudOptimisedDownloadSize(uuid, parameters, - () -> session.probeClient(Map.of( - "message", "Estimating download size...", - "timestamp", System.currentTimeMillis() - ))); + String estimateJson = dasService.estimateCloudOptimisedDownloadSize( + uuid, parameters, session::probeClient); session.send(SseEventName.ESTIMATE_COMPLETE, estimateJson); } catch (Exception e) { SseClientGoneException clientGone = SseClientGoneException.find(e); @@ -351,7 +363,10 @@ public SseEmitter estimateCloudOptimisedDownloadWithSse(String uuid, // socket left to report anything on. throw clientGone; } - log.warn("Cloud-optimised size estimation failed for UUID {}: {}", uuid, e.getMessage()); + // ERROR level, like the shared handler's upstream branch, so New Relic can alert + // on a failing data-access-service instead of it only showing up as a silent + // estimate in the portal. + log.error("Cloud-optimised size estimation failed for UUID {}", uuid, e); session.send(SseEventName.ESTIMATE_FAILED, Map.of( "message", "Size estimation failed: " + e.getMessage(), "timestamp", System.currentTimeMillis() 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 index 307ab71a..2a279f5b 100644 --- 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 @@ -57,6 +57,16 @@ public RecordingSseConnector respondWith(HttpStatusCode status) { return this; } + /** + * Answer with a status and this body, the way a gateway answers on behalf of a DAS that is + * down: the body is its own error page, not anything DAS wrote. + */ + public RecordingSseConnector respondWith(HttpStatusCode status, String body) { + this.status = status; + this.frames = List.of(body); + return this; + } + public HttpMethod method() { return lastRequest.getMethod(); } 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 index 9e4ed3e1..2ef46140 100644 --- 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 @@ -213,6 +213,71 @@ public void testNon2xxPropagatesBeforeAnyFrameIsRead() { + e.getMessage()); } + /** + * The message goes to the browser as well as the log, so a gateway's HTML error page is + * dropped: it says nothing the status has not already said, and a page of markup in the + * portal's error message helps nobody. + */ + @Test + public void testGatewayErrorPageIsReducedToTheStatus() { + connector.respondWith(HttpStatus.GATEWAY_TIMEOUT, """ + + ERROR: The request could not be satisfied +

504 Gateway Timeout ERROR

+ Generated by cloudfront (CloudFront) + """); + + RuntimeException e = assertThrows(RuntimeException.class, + () -> estimate(DasSseFrames.FrameCallback.IGNORE)); + + assertEquals("data-access-service returned 504 Gateway Timeout", e.getMessage()); + } + + /** + * What DAS itself refuses before its endpoint runs: FastAPI answers with {"detail": ...}, + * and that detail is the only thing saying which check failed, so it is unwrapped and kept. + */ + @Test + public void testFastApiDetailIsUnwrapped() { + connector.respondWith(HttpStatus.UNAUTHORIZED, "{\"detail\":\"Invalid API Key\"}"); + + RuntimeException e = assertThrows(RuntimeException.class, + () -> estimate(DasSseFrames.FrameCallback.IGNORE)); + + assertEquals("data-access-service returned 401 Unauthorized: Invalid API Key", e.getMessage()); + } + + /** + * A detail that is not a string, the shape FastAPI uses when it cannot validate a request, + * is kept as it is rather than dropped: it still says which field was rejected. + */ + @Test + public void testStructuredFastApiDetailIsKept() { + connector.respondWith(HttpStatus.UNPROCESSABLE_ENTITY, + "{\"detail\":[{\"loc\":[\"body\",\"key\"],\"msg\":\"field required\"}]}"); + + RuntimeException e = assertThrows(RuntimeException.class, + () -> estimate(DasSseFrames.FrameCallback.IGNORE)); + + assertTrue(e.getMessage().contains("field required"), "Got: " + e.getMessage()); + } + + /** + * A body in neither shape is quoted, but only as much of it as is worth reading. + */ + @Test + public void testAnOverlongBodyIsTruncated() { + connector.respondWith(HttpStatus.BAD_GATEWAY, "upstream said: " + "x".repeat(500)); + + RuntimeException e = assertThrows(RuntimeException.class, + () -> estimate(DasSseFrames.FrameCallback.IGNORE)); + + assertTrue(e.getMessage().startsWith("data-access-service returned 502 Bad Gateway: upstream said: x"), + "Got: " + e.getMessage()); + assertTrue(e.getMessage().endsWith("..."), "Got: " + e.getMessage()); + assertTrue(e.getMessage().length() < 300, "Got a message of " + e.getMessage().length() + " chars"); + } + /** * At socket level: a real DAS-shaped server that keeps * heartbeating, a client that goes away, and the assertion that the connection is dropped diff --git a/server/src/test/java/au/org/aodn/ogcapi/server/core/service/sse/SseSessionTest.java b/server/src/test/java/au/org/aodn/ogcapi/server/core/service/sse/SseSessionTest.java new file mode 100644 index 00000000..1b54ecf1 --- /dev/null +++ b/server/src/test/java/au/org/aodn/ogcapi/server/core/service/sse/SseSessionTest.java @@ -0,0 +1,167 @@ +package au.org.aodn.ogcapi.server.core.service.sse; + +import au.org.aodn.ogcapi.server.core.exception.SseClientGoneException; +import au.org.aodn.ogcapi.server.core.model.enumeration.SseEventName; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; +import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.stream.Collectors; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.mock; + +/** + * The keep-alive ticker, which is what stops an idle timeout in front of this service dropping a + * stream that is waiting on a slow upstream server, and the probe it shares the stream with. + */ +public class SseSessionTest { + + private final List written = Collections.synchronizedList(new ArrayList<>()); + + private SseSession sessionWritingTo(List events) { + SseEmitter emitter = mock(SseEmitter.class); + try { + doAnswer(invocation -> { + events.add(render(invocation.getArgument(0))); + return null; + }).when(emitter).send(any(SseEmitter.SseEventBuilder.class)); + } catch (Exception e) { + throw new IllegalStateException(e); + } + return new SseSession("test-uuid", emitter); + } + + private static String render(SseEmitter.SseEventBuilder builder) { + return builder.build().stream() + .map(part -> String.valueOf(part.getData())) + .collect(Collectors.joining()); + } + + /** + * Only what the browser would surface: a named keep-alive event, not a probe's comment line. + */ + private long keepAliveEvents() { + return List.copyOf(written).stream().filter(event -> event.contains("event:keep-alive")).count(); + } + + private long probes() { + return List.copyOf(written).stream().filter(event -> event.startsWith(":probe")).count(); + } + + /** + * Work that already writes events to its client should not have the ticker's events on top of + * its own. The connection is busy either way, which is the whole point of the ticker. + */ + @Test + @Timeout(20) + public void testTicksAreSkippedWhileTheWorkIsWritingItsOwnEvents() throws Exception { + SseSession session = sessionWritingTo(written); + session.startKeepAlive(1, () -> Map.of("message", "waiting")); + + try { + // Six events of our own, 250ms apart, so the tick at one second finds a stream that + // was written to a moment ago. + for (int i = 0; i < 6; i++) { + session.send(SseEventName.CONNECTION_ESTABLISHED, Map.of("n", i)); + Thread.sleep(250); + } + + assertEquals(0, keepAliveEvents(), + "A stream its own work is writing to does not need the ticker: " + written); + + // Nothing else is written from here, so the ticker takes over. + long deadline = System.currentTimeMillis() + 5000; + while (keepAliveEvents() == 0 && System.currentTimeMillis() < deadline) { + Thread.sleep(100); + } + + assertTrue(keepAliveEvents() > 0, "A stream that has gone quiet must be kept alive: " + written); + } finally { + session.cleanup(); + } + } + + /** + * The other half: when the ticker is the only thing writing, every tick must send. Skipping + * on "something was sent within the interval" would have skipped every second tick, since + * the ticker's own send lands just after the tick it belongs to. + */ + @Test + @Timeout(20) + public void testEveryTickSendsWhenTheTickerIsTheOnlyWriter() throws Exception { + SseSession session = sessionWritingTo(written); + session.startKeepAlive(1, () -> Map.of("message", "waiting")); + + try { + long deadline = System.currentTimeMillis() + 8000; + while (keepAliveEvents() < 3 && System.currentTimeMillis() < deadline) { + Thread.sleep(100); + } + + assertTrue(keepAliveEvents() >= 3, + "Three ticks in about three seconds, got " + keepAliveEvents() + ": " + written); + } finally { + session.cleanup(); + } + } + + /** + * The regression this guards: probing on a schedule of its own, which is what the + * cloud-optimised estimate does with the data-access-service heartbeat, used to send a + * keep-alive event. A probe landing just after a tick is too old by the next tick to skip it, + * so the two never got out of each other's way and the browser saw every keep-alive twice. + * A probe writes a comment now, so it cannot be mistaken for an event and cannot silence the + * ticker: the client reads one keep-alive per interval, no more and no fewer. + */ + @Test + @Timeout(20) + public void testProbesAreInvisibleToTheClientAndDoNotDisturbTheTicker() throws Exception { + SseSession session = sessionWritingTo(written); + session.startKeepAlive(1, () -> Map.of("message", "waiting")); + + try { + // Probe at the rate the ticker runs at, the phase that used to duplicate: DAS + // heartbeats every interval, each arriving a moment after a tick. + Thread.sleep(1100); + for (int i = 0; i < 3; i++) { + session.probeClient(); + Thread.sleep(1000); + } + + assertEquals(3, probes(), "Every probe should reach the client: " + written); + assertTrue(keepAliveEvents() >= 3, + "Probes must not silence the ticker, got " + keepAliveEvents() + ": " + written); + assertTrue(keepAliveEvents() <= 5, + "Probes must not add keep-alive events of their own, got " + keepAliveEvents() + + ": " + written); + } finally { + session.cleanup(); + } + } + + /** + * What the probe is for: the write is the only way to learn the client has gone, so a broken + * pipe has to come back as the disconnect it is rather than as a plain IOException. + */ + @Test + public void testProbeReportsABrokenPipeAsAGoneClient() throws Exception { + SseEmitter emitter = mock(SseEmitter.class); + doThrow(new IOException("Broken pipe")).when(emitter).send(any(SseEmitter.SseEventBuilder.class)); + SseSession session = new SseSession("test-uuid", emitter); + + SseClientGoneException e = assertThrows(SseClientGoneException.class, session::probeClient); + + assertTrue(e.getMessage().contains("test-uuid"), "Got: " + e.getMessage()); + } +} 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 babbf7ad..d97ea39d 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 @@ -6,6 +6,7 @@ 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.DasSseFrames; import au.org.aodn.ogcapi.server.core.util.TestLogAppender; import com.fasterxml.jackson.databind.ObjectMapper; import org.apache.logging.log4j.Level; @@ -28,6 +29,8 @@ import java.math.BigInteger; import java.util.HashMap; import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; import java.util.stream.Collectors; import static org.junit.jupiter.api.Assertions.*; @@ -57,9 +60,11 @@ public class RestApiSseTest { private final ObjectMapper objectMapper = new ObjectMapper(); + private RestServices restServices; + @BeforeEach public void setUp() { - RestServices restServices = new RestServices(batchClient, objectMapper, "test-job-definition", "test-job-queue"); + restServices = new RestServices(batchClient, objectMapper, "test-job-definition", "test-job-queue"); ReflectionTestUtils.setField(restServices, "downloadWfsDataService", downloadWfsDataService); ReflectionTestUtils.setField(restServices, "dasService", dasService); @@ -90,7 +95,7 @@ private String awaitContent(MockHttpServletResponse response, String expectedMar String content = response.getContentAsString(); while (System.currentTimeMillis() < deadline && (!content.contains(expectedMarker) - || !content.substring(content.indexOf(expectedMarker)).contains("\n\n"))) { + || !content.substring(content.indexOf(expectedMarker)).contains("\n\n"))) { Thread.sleep(50); content = response.getContentAsString(); } @@ -177,6 +182,84 @@ public void testEstimateCODownloadDasFailureEmitsEstimateFailed() throws Excepti assertTrue(content.contains("das returned 404"), "Failure reason should be forwarded in: " + content); } + /** + * The regression this guards: a data-access-service that fails without ever opening its + * stream sends no heartbeat, so with nothing else writing to the client the connection goes + * quiet for as long as that failure takes, 30-60s for a gateway timeout. + */ + @Test + public void testEstimateCOKeepsTheClientAliveWhileDasSaysNothing() throws Exception { + ReflectionTestUtils.setField(restServices, "estimateKeepAliveSeconds", 1L); + + CountDownLatch dasAnswers = new CountDownLatch(1); + when(dasService.estimateCloudOptimisedDownloadSize(any(), anyMap(), any())) + .thenAnswer(invocation -> { + // No heartbeat and no result, the way a gateway 504 for a downed DAS arrives. + assertTrue(dasAnswers.await(5, TimeUnit.SECONDS), "Test did not release the DAS call"); + throw new RuntimeException("data-access-service returned 504 Gateway Timeout"); + }); + + 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 whileWaiting = awaitContent(response, "event:keep-alive"); + assertEventOrder(whileWaiting, "connection-established", "keep-alive"); + assertFalse(whileWaiting.contains("event:estimate-failed"), + "The estimate has not failed yet: " + whileWaiting); + + dasAnswers.countDown(); + + String content = awaitContent(response, "event:estimate-failed"); + assertEventOrder(content, "keep-alive", "estimate-failed"); + assertTrue(content.contains("504 Gateway Timeout"), "Failure reason should be forwarded in: " + content); + } + + /** + * The regression this guards: every data-access-service heartbeat is probed on, and a probe + * used to be a keep-alive event. DAS heartbeats at about the rate the ticker runs at, so the + * client received the ticker's keep-alive and a heartbeat's a moment later and read the same + * event twice. A probe writes an SSE comment now, which EventSource discards, so heartbeats + * keep the connection busy without reaching the client at all. + */ + @Test + public void testEstimateCODasHeartbeatsDoNotReachTheClientAsEvents() throws Exception { + // The ticker is left at its default interval, longer than this test runs, so every + // keep-alive in the response would have to have come from a heartbeat. + when(dasService.estimateCloudOptimisedDownloadSize(any(), anyMap(), any())) + .thenAnswer(invocation -> { + DasSseFrames.FrameCallback onHeartbeat = invocation.getArgument(2); + for (int i = 0; i < 3; i++) { + onHeartbeat.onFrame(); + } + return "{\"estimated_output_bytes\":12345}"; + }); + + 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 content = awaitContent(response, "event:estimate-complete"); + + assertFalse(content.contains("event:keep-alive"), + "A heartbeat must not surface as an event the client reads: " + content); + assertEquals(3, countOf(content, ":probe"), + "Each heartbeat should still write to the client: " + content); + } + + private static int countOf(String content, String needle) { + int count = 0; + for (int at = content.indexOf(needle); at >= 0; at = content.indexOf(needle, at + needle.length())) { + count++; + } + return count; + } + @Test public void testEstimateCODownloadClientDisconnectIsNotReportedAsAFailedEstimate() throws Exception { // A disconnect is what aborts the DAS call, and the estimate has an unchecked signature,