limit concurrent downloads per user and queue the rest

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

@@ -9,6 +9,7 @@
 import au.org.aodn.ogcapi.server.core.util.GeometryUtils;
 import au.org.aodn.ogcapi.server.core.util.RestTemplateUtils;
 import au.org.aodn.ogcapi.server.processes.BatchJobProperties;
+import au.org.aodn.ogcapi.server.processes.DownloadLimitProperties;
 import com.fasterxml.jackson.annotation.JsonInclude;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import jakarta.annotation.PostConstruct;
@@ -35,6 +36,7 @@
         GNProperties.class,
         DasProperties.class,
         BatchJobProperties.class,
+        DownloadLimitProperties.class,
         OgcApiProperties.class
 })
 public class Config {

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

@@ -0,0 +1,8 @@
+package au.org.aodn.ogcapi.server.core.exception;
+
+public class DownloadLimitExceededException extends RuntimeException {
+    public DownloadLimitExceededException(int maxConcurrent) {
+        super("You already have " + maxConcurrent + " downloads in progress. "
+                + "Wait for one of them to complete before starting another.");
+    }
+}

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

@@ -116,6 +116,20 @@ public ResponseEntity<ErrorResponse> handleDownloadJobStatusException(
         return new ResponseEntity<>(errorResponse, HttpStatus.INTERNAL_SERVER_ERROR);
     }
 
+    @ExceptionHandler(DownloadLimitExceededException.class)
+    public ResponseEntity<ErrorResponse> handleDownloadLimitExceededException(
+            DownloadLimitExceededException ex,
+            WebRequest request) {
+        ErrorResponse errorResponse = ErrorResponse
+                .builder()
+                .timestamp(LocalDateTime.now())
+                .message(ex.getMessage())
+                .details(request.getDescription(false))
+                .build();
+
+        return new ResponseEntity<>(errorResponse, HttpStatus.TOO_MANY_REQUESTS);
+    }
+
     @ExceptionHandler(DasUpstreamException.class)
     public ResponseEntity<ErrorResponse> handleDasUpstreamException(DasUpstreamException ex, WebRequest request) {
         ErrorResponse errorResponse = ErrorResponse

server/src/main/java/au/org/aodn/ogcapi/server/core/model/DownloadExecutionResponse.java

@@ -4,7 +4,7 @@
 import com.fasterxml.jackson.annotation.JsonProperty;
 import io.swagger.v3.oas.annotations.media.Schema;
 
-@Schema(description = "Compatible download execution response with the submitted AWS Batch job ID.")
+@Schema(description = "Compatible download execution response with the submitted job ID.")
 public record DownloadExecutionResponse(
         @JsonProperty("message") InlineValue message,
         @JsonProperty("status") InlineValue status,

server/src/main/java/au/org/aodn/ogcapi/server/processes/DownloadAdmissionService.java

@@ -0,0 +1,111 @@
+package au.org.aodn.ogcapi.server.processes;
+
+import au.org.aodn.ogcapi.server.core.exception.DownloadLimitExceededException;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.locks.ReentrantLock;
+
+/**
+ * Caps how many downloads one user can have running at once. A user already at the limit is
+ * rejected outright: the caller has to wait for one of their own downloads to finish and
+ * try again.
+ */
+@Slf4j
+@Service
+public class DownloadAdmissionService {
+
+    private final RestServices restServices;
+    private final InFlightDownloadCounter counter;
+    private final DownloadLimitProperties limits;
+
+    private final ReentrantLock lock = new ReentrantLock();
+
+    /**
+     * Recipients with a submit currently in flight, not yet reflected in {@link #counter}
+     * because the AWS call it is waiting on has not returned. Consulted together with the
+     * counter whenever admission checks whether a recipient has room - without it, two
+     * requests racing during the same in-flight submit could both see room and both go
+     * through, letting a user briefly exceed the limit. Guarded by {@link #lock}.
+     */
+    private final Map<String, Integer> reserved = new HashMap<>();
+
+    @Autowired
+    public DownloadAdmissionService(
+            RestServices restServices,
+            InFlightDownloadCounter counter,
+            DownloadLimitProperties limits) {
+        this.restServices = restServices;
+        this.counter = counter;
+        this.limits = limits;
+    }
+
+    /**
+     * Submit a download to AWS Batch and return its job id.
+     *
+     * @throws DownloadLimitExceededException the recipient already has {@code maxConcurrent}
+     *                                         downloads running
+     */
+    public String submit(DownloadRequest request) throws JsonProcessingException {
+        String key = InFlightDownloadCounter.recipientKey(request.recipient());
+
+        if (limits.enabled()) {
+            // Outside the lock: the sweep is the only expensive step.
+            counter.refreshIfStale();
+            reserveOrReject(request, key);
+        }
+
+        try {
+            Map<String, String> parameters = restServices.buildDownloadParameters(request);
+            String jobName = RestServices.downloadJobName(request.recipient());
+            String awsJobId = restServices.submitDownloadJob(jobName, parameters);
+            counter.recordSubmitted(awsJobId, request.recipient());
+            notifyStarted(request);
+            return awsJobId;
+        } finally {
+            if (limits.enabled()) {
+                releaseReservation(key);
+            }
+        }
+    }
+
+    private void reserveOrReject(DownloadRequest request, String key) {
+        lock.lock();
+        try {
+            int inFlight = counter.countInFlight(request.recipient()) + reserved.getOrDefault(key, 0);
+            if (inFlight >= limits.maxConcurrent()) {
+                throw new DownloadLimitExceededException(limits.maxConcurrent());
+            }
+            reserved.merge(key, 1, Integer::sum);
+        } finally {
+            lock.unlock();
+        }
+    }
+
+    private void releaseReservation(String key) {
+        lock.lock();
+        try {
+            reserved.computeIfPresent(key, (k, count) -> count <= 1 ? null : count - 1);
+        } finally {
+            lock.unlock();
+        }
+    }
+
+    private void notifyStarted(DownloadRequest request) {
+        restServices.notifyUser(
+                request.recipient(),
+                request.uuid(),
+                request.key(),
+                request.startDate(),
+                request.endDate(),
+                request.multiPolygon(),
+                request.collectionTitle(),
+                request.fullMetadataLink(),
+                request.suggestedCitation(),
+                request.outputFormat());
+    }
+}

server/src/main/java/au/org/aodn/ogcapi/server/processes/DownloadJobStatusService.java

@@ -37,6 +37,10 @@ public class DownloadJobStatusService {
 
     static final String PROCESS_ID = "download-dataset";
     static final Duration CHILD_DISCOVERY_WINDOW = Duration.ofSeconds(30);
+    // These exact names are an internal contract with data-access-service. Any DAS
+    // naming change must be applied here at the same time.
+    static final String PREPARE_NAME_PREFIX = "prepare-data-for-job-";
+    static final String COLLECT_NAME_PREFIX = "collect-data-for-job-";
 
     private static final String INITIAL_TYPE = "sub-setting";
     private static final String PREPARE_TYPE = "sub-setting-data-preparation";
@@ -71,7 +75,6 @@ public DownloadJobStatusService(
 
     public DownloadJobStatusInfo getStatus(String jobId) {
         validateJobId(jobId);
-
         try {
             JobDetail initial = describeInitialJob(jobId);
 
@@ -83,12 +86,10 @@ public DownloadJobStatusInfo getStatus(String jobId) {
                 return toStatusInfo(jobId, status, initial, null, null);
             }
 
-            // These exact names are an internal contract with data-access-service. Any DAS
-            // naming change must be applied here at the same time.
             JobDetail prepare = findChildJob(
-                    "prepare-data-for-job-" + jobId, jobId, PREPARE_TYPE);
+                    PREPARE_NAME_PREFIX + jobId, jobId, PREPARE_TYPE);
             JobDetail collect = findChildJob(
-                    "collect-data-for-job-" + jobId, jobId, COLLECT_TYPE);
+                    COLLECT_NAME_PREFIX + jobId, jobId, COLLECT_TYPE);
 
             boolean discoveryWindowExpired = discoveryWindowExpired(initial);
             DownloadJobStatusAggregator.WorkflowMode workflowMode = isExplicitZarr(initial.parameters())

server/src/main/java/au/org/aodn/ogcapi/server/processes/DownloadLimitProperties.java

@@ -0,0 +1,20 @@
+package au.org.aodn.ogcapi.server.processes;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.boot.context.properties.bind.DefaultValue;
+
+import java.time.Duration;
+
+/**
+ * Per-user concurrency limit for dataset downloads. A user, identified by the recipient
+ * email, may have at most {@code maxConcurrent} downloads in flight at once; a request past
+ * that is rejected outright.
+ */
+@ConfigurationProperties(prefix = "aws.batch.job.user-limit")
+public record DownloadLimitProperties(
+        @DefaultValue("true") boolean enabled,
+        @DefaultValue("10") int maxConcurrent,
+        /** How long the shared in-flight snapshot is reused before the queues are swept again. */
+        @DefaultValue("15s") Duration refreshInterval
+) {
+}

server/src/main/java/au/org/aodn/ogcapi/server/processes/DownloadRequest.java

@@ -0,0 +1,20 @@
+package au.org.aodn.ogcapi.server.processes;
+
+/**
+ * The inputs of one {@code download} execute request, as extracted from the OGC Execute
+ * body. Carried as a unit so a request that has to wait for a free slot can be submitted
+ * later exactly as it arrived.
+ */
+public record DownloadRequest(
+        String uuid,
+        String key,
+        String startDate,
+        String endDate,
+        Object multiPolygon,
+        String recipient,
+        String collectionTitle,
+        String fullMetadataLink,
+        String suggestedCitation,
+        String outputFormat
+) {
+}