keep estimate stream alive while DAS is silent

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<DatasetMetadata> getDatasetMetadata(String datasetId) {

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<ScheduledFuture<?>> keepAliveTaskRef = new AtomicReference<>();
     private final AtomicReference<ScheduledExecutorService> 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<Object> 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

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()

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();
     }

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, """
+                <!DOCTYPE HTML PUBLIC "-//W3C//DTD HTML 4.01 Transitional//EN">
+                <HTML><HEAD><TITLE>ERROR: The request could not be satisfied</TITLE></HEAD>
+                <BODY><H1>504 Gateway Timeout ERROR</H1>
+                Generated by cloudfront (CloudFront)</BODY></HTML>
+                """);
+
+        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