add caching config for co estimation

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

@@ -35,6 +35,7 @@ public class CacheConfig {
 
     public static final String DOWNLOADABLE_FIELDS = "downloadable-fields";
     public static final String DOWNLOADABLE_SIZE = "downloadable-size";
+    public static final String CLOUD_OPTIMISED_ESTIMATE = "cloud-optimised-estimate";
 
     public static final String ALL_NO_LAND_GEOMETRY = "all-noland-geometry";
     public static final String ALL_PARAM_VOCABS = "parameter-vocabs";
@@ -91,6 +92,12 @@ public JCacheCacheManager cacheManager() throws IOException {
                                 ResourcePoolsBuilder.heap(100)
                         ).withExpiry(ExpiryPolicyBuilder.timeToLiveExpiration(Duration.ofHours(24)))
                 )
+                .withCache(CLOUD_OPTIMISED_ESTIMATE,
+                        CacheConfigurationBuilder.newCacheConfigurationBuilder(
+                                Object.class, String.class,
+                                ResourcePoolsBuilder.heap(200)
+                        ).withExpiry(ExpiryPolicyBuilder.timeToLiveExpiration(Duration.ofHours(24)))
+                )
                 .withCache(ELASTIC_SEARCH_UUID_ONLY,
                         CacheConfigurationBuilder.newCacheConfigurationBuilder(
                                 Object.class, Object.class,

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

@@ -1,12 +1,14 @@
 package au.org.aodn.ogcapi.server.core.service.das;
 
+import au.org.aodn.ogcapi.server.core.configuration.CacheConfig;
 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.DasSseFrames;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.cache.annotation.Cacheable;
 import org.springframework.core.ParameterizedTypeReference;
 import org.springframework.http.*;
 import org.springframework.http.codec.ServerSentEvent;
@@ -108,7 +110,7 @@ public ResponseEntity<byte[]> getMooringDetailsBetweenDates(String startDateTime
     /**
      * 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:
+     * sends (see SubsetParametersUtils), so DAS treats both alike. Four 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
@@ -120,7 +122,14 @@ public ResponseEntity<byte[]> getMooringDetailsBetweenDates(String startDateTime
      * 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.
+     * 4. The result is cached for 24 hours under uuid plus parameters, so repeating the same
+     * subset answers without calling DAS at all. Only the final result is ever cached: heartbeats
+     * are handed to onHeartbeat as they arrive and never become a return value. onHeartbeat is
+     * left out of the key on purpose, because it is a new lambda per request and including it
+     * would make every key unique; on a hit the body never runs, so there is nothing to keep
+     * alive anyway. A failed estimate throws, and a throwing call caches nothing.
      */
+    @Cacheable(cacheNames = CacheConfig.CLOUD_OPTIMISED_ESTIMATE, key = "{#uuid, #parameters}")
     public String estimateCloudOptimisedDownloadSize(String uuid,
                                                      Map<String, String> parameters,
                                                      DasSseFrames.FrameCallback onHeartbeat) {

server/src/test/java/au/org/aodn/ogcapi/server/core/http/RecordingSseConnector.java

@@ -38,6 +38,7 @@ public final class RecordingSseConnector implements ClientHttpConnector {
     private String lastBody;
     private final AtomicBoolean cancelled = new AtomicBoolean();
     private final AtomicInteger framesDelivered = new AtomicInteger();
+    private final AtomicInteger requests = new AtomicInteger();
 
     /**
      * Answer with 200 and these SSE frames, each one a complete frame ending in a blank line.
@@ -91,12 +92,21 @@ public int framesDelivered() {
         return framesDelivered.get();
     }
 
+    /**
+     * How many times a WebClient asked to connect, so a test can tell a real call from a
+     * cached one.
+     */
+    public int requests() {
+        return requests.get();
+    }
+
     @Override
     public Mono<ClientHttpResponse> connect(HttpMethod method, URI uri,
                                             Function<? super ClientHttpRequest, Mono<Void>> requestCallback) {
 
         MockClientHttpRequest request = new MockClientHttpRequest(method, uri);
         lastRequest = request;
+        requests.incrementAndGet();
 
         return requestCallback.apply(request)
                 // Deferred: the mock only has a body once the callback above has written one.

server/src/test/java/au/org/aodn/ogcapi/server/core/service/das/DasServiceCacheTest.java

@@ -0,0 +1,191 @@
+package au.org.aodn.ogcapi.server.core.service.das;
+
+import au.org.aodn.ogcapi.server.core.configuration.CacheConfig;
+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.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.springframework.cache.CacheManager;
+import org.springframework.cache.annotation.EnableCaching;
+import org.springframework.cache.concurrent.ConcurrentMapCacheManager;
+import org.springframework.context.annotation.AnnotationConfigApplicationContext;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.web.client.RestTemplate;
+import org.springframework.web.reactive.function.client.WebClient;
+
+import java.time.Duration;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+/**
+ * Covers the cache in front of the cloud-optimised size estimate: a repeat of the same subset is
+ * answered without calling DAS, a different subset is not, and a failure leaves nothing behind.
+ * The other DAS tests build the service with new, which means no proxy and no caching, so caching
+ * only shows up with a Spring context around the bean. What is under test here is the annotation
+ * and its key, not EhCache, so a plain ConcurrentMapCacheManager stands in for the real one.
+ */
+public class DasServiceCacheTest {
+
+    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 static final String OTHER_RESULT_FRAME = """
+            event: result
+            data: {"status":"completed","data":{"estimated_output_bytes":456}}
+
+            """;
+
+    private static final String ERROR_FRAME = """
+            event: error
+            data: {"status":"error","message":"boom"}
+
+            """;
+
+    /**
+     * proxyTargetClass matches what Spring Boot does by default. DasService implements
+     * ApplicationInfo, so a JDK proxy would only expose that interface and the service could not
+     * be looked up, or injected into RestServices, by its own type.
+     */
+    @Configuration
+    @EnableCaching(proxyTargetClass = true)
+    static class CachingContext {
+
+        @Bean
+        public CacheManager cacheManager() {
+            return new ConcurrentMapCacheManager(CacheConfig.CLOUD_OPTIMISED_ESTIMATE);
+        }
+
+        @Bean
+        public RecordingSseConnector connector() {
+            return new RecordingSseConnector();
+        }
+
+        @Bean
+        public DasService dasService(RecordingSseConnector connector) {
+            // A null infoPath keeps the constructor's info query from making a request.
+            DasProperties properties = new DasProperties(HOST, null, "test-secret", null,
+                    Duration.ofSeconds(5), Duration.ofSeconds(30), Duration.ofMinutes(2));
+            WebClient webClient = WebClient.builder()
+                    .clientConnector(connector)
+                    .baseUrl(HOST)
+                    .build();
+            return new DasService(properties, new RestTemplate(), webClient, new ObjectMapper());
+        }
+    }
+
+    private AnnotationConfigApplicationContext context;
+    private DasService dasService;
+    private RecordingSseConnector connector;
+
+    @BeforeEach
+    public void setUp() {
+        context = new AnnotationConfigApplicationContext(CachingContext.class);
+        dasService = context.getBean(DasService.class);
+        connector = context.getBean(RecordingSseConnector.class);
+    }
+
+    @AfterEach
+    public void tearDown() {
+        context.close();
+    }
+
+    /**
+     * A new map each time, the way SubsetParametersUtils builds one per request, so the tests
+     * show the key matching on content rather than on the same instance coming back.
+     */
+    private static Map<String, String> parameters(String outputFormat) {
+        Map<String, String> parameters = new HashMap<>();
+        parameters.put("uuid", "test-uuid");
+        parameters.put("key", "a.zarr");
+        parameters.put("start_date", "2020-01-01");
+        parameters.put("end_date", "2020-12-31");
+        parameters.put("multi_polygon", "non-specified");
+        parameters.put("output_format", outputFormat);
+        return parameters;
+    }
+
+    /**
+     * A distinct callback instance each call, the way RestServices passes a new
+     * session::probeClient per request. Reusing one constant here would hide a key that wrongly
+     * included the callback, because that key would still match on the second call.
+     */
+    private static DasSseFrames.FrameCallback freshCallback() {
+        return new AtomicInteger()::incrementAndGet;
+    }
+
+    private String estimate(Map<String, String> parameters, DasSseFrames.FrameCallback onHeartbeat) {
+        return dasService.estimateCloudOptimisedDownloadSize("test-uuid", parameters, onHeartbeat);
+    }
+
+    @Test
+    public void testRepeatedEstimateIsServedFromCache() {
+        connector.respondWith(List.of(RESULT_FRAME));
+
+        String first = estimate(parameters("netcdf"), freshCallback());
+        String second = estimate(parameters("netcdf"), freshCallback());
+
+        assertEquals("{\"estimated_output_bytes\":123}", first);
+        assertEquals(first, second, "The repeat returns the same estimate");
+        assertEquals(1, connector.requests(), "The repeat is answered from the cache, not by data-access-service");
+    }
+
+    @Test
+    public void testDifferentParametersStillCallDas() {
+        connector.respondWith(List.of(RESULT_FRAME));
+        estimate(parameters("netcdf"), freshCallback());
+
+        connector.respondWith(List.of(OTHER_RESULT_FRAME));
+        String second = estimate(parameters("csv"), freshCallback());
+
+        assertEquals("{\"estimated_output_bytes\":456}", second);
+        assertEquals(2, connector.requests(), "A different output format is a different subset, so a different key");
+    }
+
+    @Test
+    public void testFailedEstimateIsNotCached() {
+        connector.respondWith(List.of(ERROR_FRAME));
+        assertThrows(RuntimeException.class,
+                () -> estimate(parameters("netcdf"), freshCallback()));
+
+        connector.respondWith(List.of(RESULT_FRAME));
+        String retry = estimate(parameters("netcdf"), freshCallback());
+
+        assertEquals("{\"estimated_output_bytes\":123}", retry);
+        assertEquals(2, connector.requests(), "A failure caches nothing, so the retry reaches data-access-service");
+    }
+
+    /**
+     * The callback is a new lambda per request, so it has to stay out of the key. A hit that
+     * still ran the body, or a key that included the callback, would both show up here.
+     */
+    @Test
+    public void testCacheHitSkipsTheStreamAndItsHeartbeats() {
+        connector.respondWith(List.of(HEARTBEAT_FRAME, RESULT_FRAME));
+
+        AtomicInteger heartbeats = new AtomicInteger();
+        estimate(parameters("netcdf"), heartbeats::incrementAndGet);
+        assertEquals(1, heartbeats.get(), "The first estimate reads the DAS stream");
+
+        estimate(parameters("netcdf"), heartbeats::incrementAndGet);
+        assertEquals(1, heartbeats.get(), "A cache hit never opens a stream, so nothing heartbeats");
+    }
+}