Repository: aehrc/pathling (language: python)

Issue:
Improve cacheability of async requests
Currently, it is necessary to cache 202 requests on the client side in order to achieve caching of async requests. This is not ideal, as 202s are technically not cacheable and are not cached by default by many proxies and CDNs.

This will change the way that we deal with async requests within the server such that identical requests will always be routed to the same job ID number. This will remove the need to cache the initial request, and clients will continue to be able to cache the final result.


Discussion / hints:

Fix patch:
diff --git a/fhir-server/src/main/java/au/csiro/pathling/aggregate/AggregateProvider.java b/fhir-server/src/main/java/au/csiro/pathling/aggregate/AggregateProvider.java
index f41fef13d5..79ba911407 100644
--- a/fhir-server/src/main/java/au/csiro/pathling/aggregate/AggregateProvider.java
+++ b/fhir-server/src/main/java/au/csiro/pathling/aggregate/AggregateProvider.java
@@ -23,14 +23,12 @@
 import au.csiro.pathling.security.OperationAccess;
 import ca.uhn.fhir.rest.annotation.Operation;
 import ca.uhn.fhir.rest.annotation.OperationParam;
-import ca.uhn.fhir.rest.api.server.RequestDetails;
 import ca.uhn.fhir.rest.server.IResourceProvider;
+import ca.uhn.fhir.rest.server.servlet.ServletRequestDetails;
 import java.util.List;
 import java.util.Optional;
 import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
-import javax.servlet.http.HttpServletRequest;
-import javax.servlet.http.HttpServletResponse;
 import org.hl7.fhir.instance.model.api.IBaseResource;
 import org.hl7.fhir.r4.model.Enumerations.ResourceType;
 import org.hl7.fhir.r4.model.Parameters;
@@ -80,9 +78,7 @@ public Class<? extends IBaseResource> getResourceType() {
    * @param aggregation a list of aggregation expressions
    * @param grouping a list of grouping expressions
    * @param filter a list of filter expressions
-   * @param request the {@link HttpServletRequest} details
-   * @param requestDetails the {@link RequestDetails} containing HAPI inferred info
-   * @param response the {@link HttpServletResponse} response
+   * @param requestDetails the {@link ServletRequestDetails} containing HAPI inferred info
    * @return {@link Parameters} object representing the result
    */
   @Operation(name = "$aggregate", idempotent = true)
@@ -92,9 +88,7 @@ public Parameters aggregate(
       @Nullable @OperationParam(name = "aggregation") final List<String> aggregation,
       @Nullable @OperationParam(name = "grouping") final List<String> grouping,
       @Nullable @OperationParam(name = "filter") final List<String> filter,
-      @SuppressWarnings("unused") @Nullable final HttpServletRequest request,
-      @SuppressWarnings("unused") @Nullable final RequestDetails requestDetails,
-      @SuppressWarnings("unused") @Nullable final HttpServletResponse response) {
+      @SuppressWarnings("unused") @Nullable final ServletRequestDetails requestDetails) {
     final AggregateRequest query = AggregateRequest.fromUserInput(
         resourceType, Optional.ofNullable(aggregation), Optional.ofNullable(grouping),
         Optional.ofNullable(filter));
diff --git a/fhir-server/src/main/java/au/csiro/pathling/async/AsyncAspect.java b/fhir-server/src/main/java/au/csiro/pathling/async/AsyncAspect.java
index 56d7d2fef1..db60bfc1c7 100644
--- a/fhir-server/src/main/java/au/csiro/pathling/async/AsyncAspect.java
+++ b/fhir-server/src/main/java/au/csiro/pathling/async/AsyncAspect.java
@@ -18,19 +18,19 @@
 package au.csiro.pathling.async;
 
 import static au.csiro.pathling.security.SecurityAspect.getCurrentUserId;
-import static java.util.Objects.requireNonNull;
 
 import au.csiro.pathling.errors.DiagnosticContext;
 import au.csiro.pathling.errors.ErrorHandlingInterceptor;
 import au.csiro.pathling.errors.ErrorReportingInterceptor;
-import ca.uhn.fhir.rest.api.server.RequestDetails;
 import ca.uhn.fhir.rest.server.exceptions.BaseServerResponseException;
+import ca.uhn.fhir.rest.server.servlet.ServletRequestDetails;
 import java.lang.reflect.UndeclaredThrowableException;
 import java.util.Arrays;
 import java.util.List;
+import java.util.Map;
 import java.util.Map.Entry;
-import java.util.Objects;
 import java.util.Optional;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.Future;
 import java.util.stream.Collectors;
 import javax.annotation.Nonnull;
@@ -77,6 +77,12 @@ public class AsyncAspect {
   @Nonnull
   private final JobRegistry jobRegistry;
 
+  @Nonnull
+  private final RequestTagFactory requestTagFactory;
+
+  @Nonnull
+  private final Map<RequestTag, Job> requestTagToJob = new ConcurrentHashMap<>();
+
   @Nonnull
   private final StageMap stageMap;
 
@@ -85,77 +91,81 @@ public class AsyncAspect {
 
   /**
    * @param executor used to run asynchronous jobs in the background
+   * @param requestTagFactory used to create {@link RequestTag} instances
    * @param jobRegistry the {@link JobRegistry} used to keep track of running jobs
    * @param stageMap the {@link StageMap} used to map stages to job IDs
    * @param spark used for updating the Spark Context with job identity
    */
   public AsyncAspect(@Nonnull final ThreadPoolTaskExecutor executor,
+      @Nonnull final RequestTagFactory requestTagFactory,
       @Nonnull final JobRegistry jobRegistry, @Nonnull final StageMap stageMap,
       @Nonnull final SparkSession spark) {
     this.executor = executor;
+    this.requestTagFactory = requestTagFactory;
     this.jobRegistry = jobRegistry;
     this.stageMap = stageMap;
     this.spark = spark;
   }
 
   @Around("@annotation(asyncSupported)")
-  private IBaseResource maybeExecuteAsynchronously(@Nonnull final ProceedingJoinPoint joinPoint,
+  protected IBaseResource maybeExecuteAsynchronously(@Nonnull final ProceedingJoinPoint joinPoint,
       @Nonnull final AsyncSupported asyncSupported) throws Throwable {
     final Object[] args = joinPoint.getArgs();
-    final HttpServletRequest request = getRequest(args);
+    final ServletRequestDetails requestDetails = getServletRequestDetails(args);
+    final HttpServletRequest request = requestDetails.getServletRequest();
     final String prefer = request.getHeader(ASYNC_HEADER);
 
     if (prefer != null && prefer.equals(ASYNC_HEADER_VALUE)) {
       log.info("Asynchronous processing requested");
-      processRequestAsynchronously(joinPoint, args, spark);
+      processRequestAsynchronously(joinPoint, requestDetails, spark);
       throw new ProcessingNotCompletedException("Accepted", buildOperationOutcome());
     } else {
-      return (IBaseResource) joinPoint.proceed(args);
+      return (IBaseResource) joinPoint.proceed();
     }
   }
 
   private void processRequestAsynchronously(@Nonnull final ProceedingJoinPoint joinPoint,
-      @Nonnull final Object[] args, @Nonnull final SparkSession spark) {
-    final RequestDetails requestDetails = getRequestDetails(args);
-    final HttpServletResponse response = getResponse(args);
-    final String requestId = requestDetails.getRequestId();
-    final Authentication authentication = SecurityContextHolder.getContext().getAuthentication();
+      @Nonnull final ServletRequestDetails requestDetails,
+      @Nonnull final SparkSession spark) {
 
-    // Store the diagnostic context from the current Sentry scope.
-    final DiagnosticContext diagnosticContext = DiagnosticContext.fromSentryScope();
-
-    requireNonNull(requestId);
-    final String operation = requestDetails.getOperation().replaceFirst("\\$", "");
-    final Future<IBaseResource> result = executor.submit(() -> {
-      try {
-        diagnosticContext.configureScope(true);
-        SecurityContextHolder.getContext().setAuthentication(authentication);
-        spark.sparkContext().setJobGroup(requestId, requestId, true);
-        return (IBaseResource) joinPoint.proceed(args);
-      } catch (final Throwable e) {
-        // Unwrap the actual exception from the aspect proxy wrapper, if needed.
-        final Throwable actualEx = unwrapFromProxy(e);
-
-        // Apply the same processing and filtering as we do for synchronous requests.
-        final BaseServerResponseException convertedError = ErrorHandlingInterceptor.convertError(
-            actualEx);
-        ErrorReportingInterceptor.reportExceptionToSentry(convertedError);
-        if (ErrorReportingInterceptor.isReportableException(convertedError)) {
-          log.error("Unexpected exception in asynchronous execution.",
-              ErrorReportingInterceptor.getReportableError(convertedError));
-        } else {
-          log.warn("Asynchronous execution failed: {}.",
-              ErrorReportingInterceptor.getReportableError(convertedError).getMessage());
+    final Authentication authentication = SecurityContextHolder.getContext().getAuthentication();
+    final RequestTag requestTag = requestTagFactory.createTag(requestDetails, authentication);
+    final Job job = jobRegistry.getOrCreate(requestTag, jobId -> {
+      final DiagnosticContext diagnosticContext = DiagnosticContext.fromSentryScope();
+      final String operation = requestDetails.getOperation().replaceFirst("\\$", "");
+      final Future<IBaseResource> result = executor.submit(() -> {
+        try {
+          diagnosticContext.configureScope(true);
+          SecurityContextHolder.getContext().setAuthentication(authentication);
+          spark.sparkContext().setJobGroup(jobId, jobId, true);
+          return (IBaseResource) joinPoint.proceed();
+        } catch (final Throwable e) {
+          // Unwrap the actual exception from the aspect proxy wrapper, if needed.
+          final Throwable actualEx = unwrapFromProxy(e);
+
+          // Apply the same processing and filtering as we do for synchronous requests.
+          final BaseServerResponseException convertedError = ErrorHandlingInterceptor.convertError(
+              actualEx);
+          ErrorReportingInterceptor.reportExceptionToSentry(convertedError);
+          if (ErrorReportingInterceptor.isReportableException(convertedError)) {
+            log.error("Unexpected exception in asynchronous execution.",
+                ErrorReportingInterceptor.getReportableError(convertedError));
+          } else {
+            log.warn("Asynchronous execution failed: {}.",
+                ErrorReportingInterceptor.getReportableError(convertedError).getMessage());
+          }
+          throw new RuntimeException("Problem processing request asynchronously", actualEx);
+        } finally {
+          cleanUpAfterJob(spark, jobId);
         }
-        throw new RuntimeException("Problem processing request asynchronously", actualEx);
-      } finally {
-        cleanUpAfterJob(spark, requestId);
-      }
+      });
+      final Optional<String> ownerId = getCurrentUserId(authentication);
+      return new Job(jobId, operation, result, ownerId);
     });
-    final Optional<String> ownerId = getCurrentUserId(authentication);
-    jobRegistry.put(requestId, new Job(operation, result, ownerId));
+
+    final HttpServletResponse response = requestDetails.getServletResponse();
     response.setHeader("Content-Location",
-        requestDetails.getFhirServerBase() + "/$job?id=" + requestId);
+        requestDetails.getFhirServerBase() + "/$job?id=" + job.getId());
   }
 
   @Nonnull
@@ -168,24 +178,14 @@ private HttpServletRequest getRequest(@Nonnull final Object[] args) {
   }
 
   @Nonnull
-  private RequestDetails getRequestDetails(@Nonnull final Object[] args) {
-    return (RequestDetails) Arrays.stream(args)
-        .filter(a -> a instanceof RequestDetails)
+  private ServletRequestDetails getServletRequestDetails(@Nonnull final Object[] args) {
+    return (ServletRequestDetails) Arrays.stream(args)
+        .filter(a -> a instanceof ServletRequestDetails)
         .findFirst()
         .orElseThrow(() -> new IllegalArgumentException(
-            "Method annotated with @AsyncSupported must include a RequestDetails parameter"));
+            "Method annotated with @AsyncSupported must include a ServletRequestDetails parameter"));
   }
 
-  @Nonnull
-  private HttpServletResponse getResponse(@Nonnull final Object[] args) {
-    return (HttpServletResponse) Arrays.stream(args)
-        .filter(a -> a instanceof HttpServletResponse)
-        .findFirst()
-        .orElseThrow(() -> new IllegalArgumentException(
-            "Method annotated with @AsyncSupported must include a HttpServletResponse parameter"));
-  }
-
-
   private void cleanUpAfterJob(@Nonnull final SparkSession spark, @Nonnull final String requestId) {
     spark.sparkContext().clearJobGroup();
     // Clean up the stage mappings.
diff --git a/fhir-server/src/main/java/au/csiro/pathling/async/Job.java b/fhir-server/src/main/java/au/csiro/pathling/async/Job.java
index 16e84565bd..1bf1487d19 100644
--- a/fhir-server/src/main/java/au/csiro/pathling/async/Job.java
+++ b/fhir-server/src/main/java/au/csiro/pathling/async/Job.java
@@ -18,6 +18,7 @@
 package au.csiro.pathling.async;
 
 import java.util.Optional;
+import java.util.UUID;
 import java.util.concurrent.Future;
 import javax.annotation.Nonnull;
 import lombok.Getter;
@@ -33,6 +34,17 @@
 @ToString
 public class Job {
 
+  /**
+   * A marker interface for job tags. These are used to identify input parameters that for which an
+   * existing job can be used to provide the result.
+   */
+  public interface JobTag {
+
+  }
+  
+  @Nonnull
+  final String id;
+
   @Nonnull
   private final String operation;
 
@@ -51,8 +63,10 @@ public class Job {
    * @param result the {@link Future} result
    * @param ownerId the identifier of the owner of the job, if authenticated
    */
-  public Job(@Nonnull final String operation, @Nonnull final Future<IBaseResource> result,
+  public Job(@Nonnull final String id, @Nonnull final String operation,
+      @Nonnull final Future<IBaseResource> result,
       @Nonnull final Optional<String> ownerId) {
+    this.id = id;
     this.operation = operation;
     this.result = result;
     this.ownerId = ownerId;
@@ -75,5 +89,4 @@ public void incrementCompletedStages() {
   public int getProgressPercentage() {
     return (completedStages * 100) / totalStages;
   }
-
 }
diff --git a/fhir-server/src/main/java/au/csiro/pathling/async/JobProvider.java b/fhir-server/src/main/java/au/csiro/pathling/async/JobProvider.java
index 9efa31bad2..8483ab97bc 100644
--- a/fhir-server/src/main/java/au/csiro/pathling/async/JobProvider.java
+++ b/fhir-server/src/main/java/au/csiro/pathling/async/JobProvider.java
@@ -61,7 +61,10 @@
 @Slf4j
 public class JobProvider {
 
-  private static final Pattern ID_PATTERN = Pattern.compile("^\\w{1,50}$");
+
+  // regex for UUID
+  private static final Pattern ID_PATTERN = Pattern.compile(
+      "^[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$");
   private static final String PROGRESS_HEADER = "X-Progress";
 
   @Nonnull
@@ -93,6 +96,7 @@ public JobProvider(@Nonnull final ServerConfiguration configuration,
   public IBaseResource job(@Nullable @OperationParam(name = "id") final String id,
       @Nullable final HttpServletRequest request,
       @Nullable final HttpServletResponse response) {
+
     // Validate that the ID looks reasonable.
     if (id == null || !ID_PATTERN.matcher(id).matches()) {
       throw new ResourceNotFoundError("Job ID not found");
diff --git a/fhir-server/src/main/java/au/csiro/pathling/async/JobRegistry.java b/fhir-server/src/main/java/au/csiro/pathling/async/JobRegistry.java
index 671aa24270..3fbd201b17 100644
--- a/fhir-server/src/main/java/au/csiro/pathling/async/JobRegistry.java
+++ b/fhir-server/src/main/java/au/csiro/pathling/async/JobRegistry.java
@@ -17,7 +17,14 @@
 
 package au.csiro.pathling.async;
 
-import java.util.concurrent.ConcurrentHashMap;
+import au.csiro.pathling.async.Job.JobTag;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.UUID;
+import java.util.function.Function;
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.context.annotation.Profile;
 import org.springframework.stereotype.Component;
@@ -27,11 +34,51 @@
  *
  * @author John Grimes
  */
+@Slf4j
 @Component
 @Profile("server")
 @ConditionalOnProperty(prefix = "pathling", name = "async.enabled", havingValue = "true")
-public class JobRegistry extends ConcurrentHashMap<String, Job> {
+public class JobRegistry {
 
-  private static final long serialVersionUID = -1895384364583552139L;
+  private final Map<String, Job> jobsById = new HashMap<>();
+  private final Map<JobTag, Job> jobsByTags = new HashMap<>();
 
+  /**
+   * Gets the job with the given tag if it exists, or creates a new one using the given factory
+   * function that accepts the job id.
+   *
+   * @param tag the tag of the job.
+   * @param jobFactory the factory function that accepts the job id and returns a new job.
+   * @return the job.
+   */
+  @Nonnull
+  public synchronized Job getOrCreate(@Nonnull final JobTag tag,
+      @Nonnull final Function<String, Job> jobFactory) {
+
+    final Job existingJob = jobsByTags.get(tag);
+    if (existingJob != null) {
+      log.debug("Returning existing job: {} for tag: {}", existingJob.getId(), tag);
+      return existingJob;
+    } else {
+      final String jobId = UUID.randomUUID().toString();
+      final Job newJob = jobFactory.apply(jobId);
+      log.debug("Created new job: {} for tag: {}", newJob.getId(), tag);
+      assert jobId.equals(newJob.getId());
+      final Job replacedJob = jobsById.put(newJob.getId(), newJob);
+      assert replacedJob == null;
+      jobsByTags.put(tag, newJob);
+      return newJob;
+    }
+  }
+
+  /**
+   * Gets the jobs of the given id if exits or returns null otherwise.
+   *
+   * @param id the id of the job
+   * @return the job or null
+   */
+  @Nullable
+  public synchronized Job get(@Nonnull final String id) {
+    return jobsById.get(id);
+  }
 }
diff --git a/fhir-server/src/main/java/au/csiro/pathling/async/RequestTag.java b/fhir-server/src/main/java/au/csiro/pathling/async/RequestTag.java
new file mode 100644
index 0000000000..3f3eb3445d
--- /dev/null
+++ b/fhir-server/src/main/java/au/csiro/pathling/async/RequestTag.java
@@ -0,0 +1,54 @@
+/*
+ * Copyright 2023 Commonwealth Scientific and Industrial Research
+ * Organisation (CSIRO) ABN 41 687 119 230.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package au.csiro.pathling.async;
+
+
+import au.csiro.pathling.async.Job.JobTag;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import javax.annotation.Nonnull;
+import lombok.Value;
+
+/**
+ * Represents a tag which includes all aspects a request that need to be compared in order to
+ * determine if they represent identical requests. Two identical request should produce identical
+ * results.
+ */
+@Value
+public class RequestTag implements JobTag {
+
+  @Nonnull
+  String requestUrl;
+
+  @Nonnull
+  Map<String, List<String>> varyHeaders;
+
+  @Nonnull
+  Optional<String> cacheKey;
+
+  //toString() is overridden to exclude the varyHeaders field, which can contain sensitive data.
+  @Override
+  public String toString() {
+    return "RequestTag(" +
+        "requestUrl='" + requestUrl + "'" +
+        ", varyHeaders=List@" + Integer.toHexString(varyHeaders.hashCode()) +
+        ", cacheKey=" + cacheKey +
+        ')';
+  }
+}
diff --git a/fhir-server/src/main/java/au/csiro/pathling/async/RequestTagFactory.java b/fhir-server/src/main/java/au/csiro/pathling/async/RequestTagFactory.java
new file mode 100644
index 0000000000..e89c03dd51
--- /dev/null
+++ b/fhir-server/src/main/java/au/csiro/pathling/async/RequestTagFactory.java
@@ -0,0 +1,95 @@
+/*
+ * Copyright 2023 Commonwealth Scientific and Industrial Research
+ * Organisation (CSIRO) ABN 41 687 119 230.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package au.csiro.pathling.async;
+
+import static java.util.function.Predicate.not;
+
+import au.csiro.pathling.caching.Cacheable;
+import au.csiro.pathling.config.ServerConfiguration;
+import au.csiro.pathling.io.CacheableDatabase;
+import ca.uhn.fhir.rest.server.servlet.ServletRequestDetails;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+import lombok.Getter;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.context.annotation.Profile;
+import org.springframework.security.core.Authentication;
+import org.springframework.stereotype.Component;
+
+
+/**
+ * Factory for creating {@link RequestTag} objects using the key of provided cacheable object.
+ */
+@Component
+@Profile("server")
+@ConditionalOnProperty(prefix = "pathling", name = "async.enabled", havingValue = "true")
+@Getter
+public final class RequestTagFactory {
+
+  private final Cacheable state;
+  private final Set<String> salientHeaderNames;
+
+  private static Set<String> getSalientHeaderNames(
+      @Nonnull final ServerConfiguration configuration) {
+    final Set<String> excludedVaryHeaders = new HashSet<>(
+        configuration.getAsync().getVaryHeadersExcludedFromCacheKey());
+    return configuration.getHttpCaching().getVary().stream()
+        .filter(not(excludedVaryHeaders::contains))
+        .collect(Collectors.toUnmodifiableSet());
+  }
+
+  RequestTagFactory(@Nonnull final Cacheable state, @Nonnull final Set<String> salientHeaderNames) {
+    this.state = state;
+    this.salientHeaderNames = salientHeaderNames.stream().map(String::toLowerCase)
+        .collect(Collectors.toUnmodifiableSet());
+  }
+
+  /**
+   * Public constructor to use for Spring component bean creation.
+   */
+  @Autowired
+  public RequestTagFactory(@Nonnull final CacheableDatabase database,
+      @Nonnull final ServerConfiguration configuration) {
+    this(database, getSalientHeaderNames(configuration));
+  }
+
+  /**
+   * Creates a {@link RequestTag} object from the provided request details.
+   *
+   * @param requestDetails the request details
+   * @param authentication the authentication object
+   * @return the request tag
+   */
+  @Nonnull
+  public RequestTag createTag(@Nonnull final ServletRequestDetails requestDetails,
+      @Nullable final Authentication authentication) {
+    final Optional<String> currentCacheKey = state.getCacheKey();
+    final Map<String, List<String>> salientHeaders = requestDetails.getHeaders().entrySet().stream()
+        .filter(entry -> salientHeaderNames.contains(entry.getKey().toLowerCase()))
+        .collect(Collectors.toMap(Entry::getKey, Entry::getValue));
+    return new RequestTag(requestDetails.getCompleteUrl(), salientHeaders, currentCacheKey);
+  }
+}
diff --git a/fhir-server/src/main/java/au/csiro/pathling/config/AsyncConfiguration.java b/fhir-server/src/main/java/au/csiro/pathling/config/AsyncConfiguration.java
index 9d74ab4c3b..88be9077ea 100644
--- a/fhir-server/src/main/java/au/csiro/pathling/config/AsyncConfiguration.java
+++ b/fhir-server/src/main/java/au/csiro/pathling/config/AsyncConfiguration.java
@@ -19,6 +19,7 @@
 
 import javax.validation.constraints.NotNull;
 import lombok.Data;
+import java.util.List;
 
 /**
  * Represents configuration relating to asynchronous processing.
@@ -31,5 +32,13 @@ public class AsyncConfiguration {
    */
   @NotNull
   private boolean enabled;
-
+  
+  /**
+   * List of headers from exclude from the {@link HttpServerCachingConfiguration#getVary()} list for
+   * the purpose of server side caching of asynchronous requests. These are likely to include
+   * headers like 'Accept' or 'Accept-Encoding' which do not influence in any way the final result
+   * of the async request.
+   */
+  @NotNull
+  private List<String> varyHeadersExcludedFromCacheKey;
 }
diff --git a/fhir-server/src/main/java/au/csiro/pathling/extract/ExtractProvider.java b/fhir-server/src/main/java/au/csiro/pathling/extract/ExtractProvider.java
index 28b410011f..f6cff03675 100644
--- a/fhir-server/src/main/java/au/csiro/pathling/extract/ExtractProvider.java
+++ b/fhir-server/src/main/java/au/csiro/pathling/extract/ExtractProvider.java
@@ -26,14 +26,12 @@
 import ca.uhn.fhir.rest.annotation.OperationParam;
 import ca.uhn.fhir.rest.api.server.RequestDetails;
 import ca.uhn.fhir.rest.server.IResourceProvider;
+import ca.uhn.fhir.rest.server.servlet.ServletRequestDetails;
 import java.util.List;
-import java.util.Objects;
 import java.util.Optional;
 import java.util.UUID;
 import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
-import javax.servlet.http.HttpServletRequest;
-import javax.servlet.http.HttpServletResponse;
 import org.hl7.fhir.instance.model.api.IBaseResource;
 import org.hl7.fhir.r4.model.Enumerations.ResourceType;
 import org.hl7.fhir.r4.model.IntegerType;
@@ -84,9 +82,7 @@ public Class<? extends IBaseResource> getResourceType() {
    * @param column a list of column expressions
    * @param filter a list of filter expressions
    * @param limit a maximum number of rows to return
-   * @param request the {@link HttpServletRequest} details
-   * @param requestDetails the {@link RequestDetails} containing HAPI inferred info
-   * @param response the {@link HttpServletResponse} response
+   * @param requestDetails the {@link ServletRequestDetails} containing HAPI inferred info
    * @return {@link Parameters} object representing the result
    */
   @Operation(name = "$extract", idempotent = true)
@@ -95,9 +91,7 @@ public Parameters extract(
       @Nullable @OperationParam(name = "column") final List<String> column,
       @Nullable @OperationParam(name = "filter") final List<String> filter,
       @Nullable @OperationParam(name = "limit") final IntegerType limit,
-      @SuppressWarnings("unused") @Nullable final HttpServletRequest request,
-      @SuppressWarnings("unused") @Nullable final RequestDetails requestDetails,
-      @SuppressWarnings("unused") @Nullable final HttpServletResponse response) {
+      @SuppressWarnings("unused") @Nullable final ServletRequestDetails requestDetails) {
     return invoke(column, filter, limit, requestDetails);
   }
 
diff --git a/fhir-server/src/main/java/au/csiro/pathling/security/SecurityAspect.java b/fhir-server/src/main/java/au/csiro/pathling/security/SecurityAspect.java
index 24de08c64a..5dea5644ab 100644
--- a/fhir-server/src/main/java/au/csiro/pathling/security/SecurityAspect.java
+++ b/fhir-server/src/main/java/au/csiro/pathling/security/SecurityAspect.java
@@ -34,7 +34,7 @@
 import org.springframework.security.core.Authentication;
 import org.springframework.security.core.GrantedAuthority;
 import org.springframework.security.core.context.SecurityContextHolder;
-import org.springframework.security.oauth2.jwt.Jwt;
+import org.springframework.security.oauth2.jwt.JwtClaimAccessor;
 import org.springframework.stereotype.Component;
 
 
@@ -112,8 +112,8 @@ public static Optional<String> getCurrentUserId(@Nullable final Authentication a
     String subject = null;
     if (authentication != null) {
       final Object principal = authentication.getPrincipal();
-      if (principal instanceof Jwt) {
-        subject = ((Jwt) principal).getSubject();
+      if (principal instanceof JwtClaimAccessor) {
+        subject = ((JwtClaimAccessor) principal).getSubject();
       }
     }
     return Optional.ofNullable(subject);
diff --git a/fhir-server/src/main/java/au/csiro/pathling/update/ImportProvider.java b/fhir-server/src/main/java/au/csiro/pathling/update/ImportProvider.java
index 03a117e2c5..fd87b47cdf 100644
--- a/fhir-server/src/main/java/au/csiro/pathling/update/ImportProvider.java
+++ b/fhir-server/src/main/java/au/csiro/pathling/update/ImportProvider.java
@@ -22,6 +22,7 @@
 import ca.uhn.fhir.rest.annotation.Operation;
 import ca.uhn.fhir.rest.annotation.ResourceParam;
 import ca.uhn.fhir.rest.api.server.RequestDetails;
+import ca.uhn.fhir.rest.server.servlet.ServletRequestDetails;
 import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
 import javax.servlet.http.HttpServletRequest;
@@ -62,9 +63,7 @@ public ImportProvider(@Nonnull final ImportExecutor executor) {
    * once - multiple occurrences will result in the last input overwriting the previous ones.
    *
    * @param parameters A FHIR {@link Parameters} object describing the import request
-   * @param request the {@link HttpServletRequest} details
-   * @param requestDetails the {@link RequestDetails} containing HAPI inferred info
-   * @param response the {@link HttpServletResponse} response
+   * @param requestDetails the {@link ServletRequestDetails} containing HAPI inferred info
    * @return A FHIR {@link OperationOutcome} resource describing the result
    */
   @Operation(name = "$import")
@@ -72,9 +71,7 @@ public ImportProvider(@Nonnull final ImportExecutor executor) {
   @OperationAccess("import")
   @AsyncSupported
   public OperationOutcome importOperation(@ResourceParam final Parameters parameters,
-      @SuppressWarnings("unused") @Nullable final HttpServletRequest request,
-      @SuppressWarnings("unused") @Nullable final RequestDetails requestDetails,
-      @SuppressWarnings("unused") @Nullable final HttpServletResponse response) {
+      @SuppressWarnings("unused") @Nullable final ServletRequestDetails requestDetails) {
     return executor.execute(parameters);
   }
 
diff --git a/fhir-server/src/main/resources/application.yml b/fhir-server/src/main/resources/application.yml
index 4af35ffc8f..b2a677bde2 100644
--- a/fhir-server/src/main/resources/application.yml
+++ b/fhir-server/src/main/resources/application.yml
@@ -20,7 +20,7 @@ pathling:
   spark:
     # The name that Pathling will be identified as within the Spark cluster.
     appName: pathling
-    
+  
   storage:
     # The base URL at which Pathling will look for data files, and where it will save data received
     # within import requests. Can be an Amazon S3 (s3://), HDFS (hdfs://) or filesystem (file://)
@@ -76,7 +76,7 @@ pathling:
 
       # When GA4GH passport authentication is enabled, this option configures the list of endpoints
       # that are allowed to issue visas.
-      allowedVisaIssuers: []
+      allowedVisaIssuers: [ ]
 
   # This section configures HTTP caching response headers.
   httpCaching:
@@ -97,8 +97,8 @@ pathling:
   # This section configures the CORS functionality of the server.
   # For more information, see: https://developer.mozilla.org/en-US/docs/Web/HTTP/CORS
   cors:
-    allowedOrigins: []
-    allowedOriginPatterns: []
+    allowedOrigins: [ ]
+    allowedOriginPatterns: [ ]
     allowedMethods:
       - OPTIONS
       - GET
@@ -116,8 +116,18 @@ pathling:
     allowableSources:
       - "file:///usr/share/staging"
 
+  # This section configures the server's support asynchronous processing of HTTP requests.
   async:
+    # Enables asynchronous processing for those operations that support it,
+    # when explicitly requested by a HTTP client.
     enabled: true
+    
+    #  A subset of `pathling.httpCaching.vary` HTTP headers, which should 
+    #  be  excluded from determining that asynchronous requests are the equivalent
+    #  and can be routed to the same asynchronous job.
+    varyHeadersExcludedFromCacheKey:
+      - Accept
+      - Accept-Encoding
 
 # Use this section to set or override any Spark configuration parameter. Tuning these parameters is
 # essential to get the optimal performance for your dataset.
diff --git a/pom.xml b/pom.xml
index 13a8345682..283fa42f95 100644
--- a/pom.xml
+++ b/pom.xml
@@ -77,10 +77,11 @@
     <pathling.scalaVersion>2.12</pathling.scalaVersion>
     <pathling.scalaFullVersion>2.12.17</pathling.scalaFullVersion>
     <pathling.deltaVersion>2.2.0</pathling.deltaVersion>
+    <!-- Hadoop is currently pinned to 3.3.4 because of this vulnerability: CVE-2022-26612 -->
     <pathling.hadoopVersion>3.3.4</pathling.hadoopVersion>
     <pathling.hapiFhirVersion>6.4.4</pathling.hapiFhirVersion>
     <pathling.springBootVersion>2.7.10</pathling.springBootVersion>
-    <pathling.springSecurityVersion>5.7.5</pathling.springSecurityVersion>
+    <pathling.springSecurityVersion>5.7.8</pathling.springSecurityVersion>
     <pathling.antlrVersion>4.9.3</pathling.antlrVersion>
     <pathling.dockerJavaVersion>3.2.13</pathling.dockerJavaVersion>
     <pathling.awsSdkVersion>1.12.384</pathling.awsSdkVersion>
diff --git a/site/docs/server/configuration.md b/site/docs/server/configuration.md
index fba06b1112..32d0eb4d14 100644
--- a/site/docs/server/configuration.md
+++ b/site/docs/server/configuration.md
@@ -43,6 +43,11 @@ Additionally, you can set any variable supported by Spring Boot, see
 - `pathling.async.enabled` - (default: `true`) Enables asynchronous processing
   for those operations that support it, when explicitly requested.
 
+- `pathling.async.varyHeadersExcludedFromCacheKey` - (default: `Accept`, 
+  `Accept-Encoding`) A subset  of `pathling.httpCaching.vary` HTTP headers, 
+  which should be excluded from determining that asynchronous requests are 
+  equivalent and can be routed to the same asynchronous job.
+
 ### Encoding
 
 - `pathling.encoding.maxNestingLevel` - (default: `3`) Controls the maximum


Test patch:
diff --git a/fhir-server/src/test/java/au/csiro/pathling/async/AsyncAspectTest.java b/fhir-server/src/test/java/au/csiro/pathling/async/AsyncAspectTest.java
new file mode 100644
index 0000000000..52f9859325
--- /dev/null
+++ b/fhir-server/src/test/java/au/csiro/pathling/async/AsyncAspectTest.java
@@ -0,0 +1,225 @@
+/*
+ * Copyright 2023 Commonwealth Scientific and Industrial Research
+ * Organisation (CSIRO) ABN 41 687 119 230.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package au.csiro.pathling.async;
+
+import static au.csiro.pathling.async.RequestTagFactoryTest.createServerConfiguration;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import au.csiro.pathling.config.ServerConfiguration;
+import au.csiro.pathling.io.CacheableDatabase;
+import au.csiro.pathling.test.SpringBootUnitTest;
+import ca.uhn.fhir.rest.server.servlet.ServletRequestDetails;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.Callable;
+import java.util.concurrent.Future;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import javax.annotation.Nonnull;
+import org.apache.spark.sql.SparkSession;
+import org.aspectj.lang.ProceedingJoinPoint;
+import org.hl7.fhir.instance.model.api.IBaseResource;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentMatchers;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.mock.mockito.MockBean;
+import org.springframework.mock.web.MockHttpServletRequest;
+import org.springframework.mock.web.MockHttpServletResponse;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
+import org.springframework.security.authentication.TestingAuthenticationToken;
+import org.springframework.security.core.context.SecurityContextHolder;
+import org.springframework.security.oauth2.jwt.JwtClaimAccessor;
+
+@SpringBootUnitTest
+public class AsyncAspectTest {
+
+  @MockBean
+  private ThreadPoolTaskExecutor threadPoolTaskExecutor;
+
+  @MockBean
+  private CacheableDatabase database;
+
+  @MockBean
+  StageMap stageMap;
+
+  @MockBean
+  ProceedingJoinPoint proceedingJoinPoint;
+
+  @Autowired
+  SparkSession spark;
+
+  private JobRegistry jobRegistry;
+  private AsyncAspect asyncAspect;
+
+  private final MockHttpServletRequest servletRequest = new MockHttpServletRequest();
+  private final MockHttpServletResponse servletResponse = new MockHttpServletResponse();
+  private ServletRequestDetails requestDetails;
+
+  private static final IBaseResource RESULT_RESOURCE = mock(IBaseResource.class);
+  private static final AsyncSupported ASYNC_SUPPORTED = mock(AsyncSupported.class);
+  private static final String FHIR_SERVER_BASE = "http://localhost:8080/fhir";
+
+  // regular expression that matches content location header
+  private static final Pattern CONTENT_LOCATION_REGEX = Pattern.compile(
+      "([^?]+)\\?id=([\\w\\-]{36})");
+
+  @BeforeEach
+  public void setUp() throws Throwable {
+
+    // Wire the asynAspects and it's dependencied
+    final ServerConfiguration serverConfiguration = createServerConfiguration(
+        List.of("Accept", "Authorization"),
+        List.of("Accept"));
+    final RequestTagFactory requestTagFactory = new RequestTagFactory(database, serverConfiguration);
+    jobRegistry = new JobRegistry();
+    asyncAspect = new AsyncAspect(threadPoolTaskExecutor, requestTagFactory, jobRegistry, stageMap,
+        spark);
+
+    // Initialise mock request and response
+    requestDetails = new ServletRequestDetails();
+    requestDetails.setServletRequest(servletRequest);
+    requestDetails.setServletResponse(servletResponse);
+    requestDetails.setFhirServerBase(FHIR_SERVER_BASE);
+    requestDetails.setCompleteUrl(FHIR_SERVER_BASE + "/Patient/$aggregate?param=value1");
+    requestDetails.setOperation("$aggregate");
+
+    final Object[] args = new Object[]{requestDetails};
+    when(proceedingJoinPoint.getArgs()).thenReturn(args);
+    when(proceedingJoinPoint.proceed()).thenReturn(RESULT_RESOURCE);
+  }
+
+  @Nonnull
+  IBaseResource executeRequest() throws Throwable {
+    return asyncAspect.maybeExecuteAsynchronously(
+        proceedingJoinPoint, ASYNC_SUPPORTED);
+  }
+
+  void setAuthenticationPrincipal(@Nonnull final Object principal) {
+    SecurityContextHolder.getContext()
+        .setAuthentication(new TestingAuthenticationToken(principal, ""));
+  }
+
+  @Nonnull
+  String assertExecutedAsync() {
+    servletRequest.removeHeader("Prefer");
+    servletRequest.addHeader("Prefer", "respond-async");
+    final ProcessingNotCompletedException ex = assertThrows(ProcessingNotCompletedException.class,
+        this::executeRequest);
+    assertEquals(202, ex.getStatusCode());
+    final String contentLocation = (String) servletResponse.getHeaderValue("Content-Location");
+    assertNotNull(contentLocation);
+    final Matcher matcher = CONTENT_LOCATION_REGEX.matcher(contentLocation);
+    assertTrue(matcher.matches());
+    final String requestUrl = matcher.group(1);
+    assertEquals(FHIR_SERVER_BASE + "/$job", requestUrl);
+    // return job id
+    return matcher.group(2);
+  }
+
+
+  @Test
+  public void testSynchronousRequestReturnsExpectedResponse() throws Throwable {
+    final IBaseResource result = executeRequest();
+    assertEquals(RESULT_RESOURCE, result);
+  }
+
+  @Test
+  @SuppressWarnings("unchecked")
+  public void testAsyncRequestsSchedulesNewJob() {
+    // setup thread pool executor to return a mock future
+    final Future<IBaseResource> mockFuture = mock(Future.class);
+    when(threadPoolTaskExecutor.submit(ArgumentMatchers.<Callable<IBaseResource>>any())).thenReturn(
+        mockFuture);
+
+    // setup authentication principal
+    final JwtClaimAccessor mockJwtPrincipal = mock(JwtClaimAccessor.class);
+    when(mockJwtPrincipal.getSubject()).thenReturn("subject1");
+    setAuthenticationPrincipal(mockJwtPrincipal);
+
+    final String jobId = assertExecutedAsync();
+    final Job newJob = jobRegistry.get(jobId);
+    assertNotNull(newJob);
+    assertEquals(jobId, newJob.getId());
+    assertEquals(mockFuture, newJob.getResult());
+    assertEquals("aggregate", newJob.getOperation());
+    assertEquals(Optional.of("subject1"), newJob.getOwnerId());
+  }
+
+  @Test
+  public void testReusesAsynJobIfOnlyWhitelistedHeadersChange() {
+    setAuthenticationPrincipal("principal1");
+    servletRequest.addHeader("Accept", "value1");
+    assertEquals("value1", servletRequest.getHeader("Accept"));
+    final String jobId1 = assertExecutedAsync();
+
+    setAuthenticationPrincipal("principal1");
+    servletRequest.removeHeader("Accept");
+    servletRequest.addHeader("Accept", "value2");
+    assertEquals("value2", servletRequest.getHeader("Accept"));
+    final String jobId2 = assertExecutedAsync();
+
+    assertEquals(jobId1, jobId2);
+  }
+
+  @Test
+  public void testCreatesNewAsyncJobWhenSalientHeaderChanges() {
+    servletRequest.addHeader("Authorization", "value1");
+    assertEquals("value1", servletRequest.getHeader("Authorization"));
+    final String jobId1 = assertExecutedAsync();
+
+    servletRequest.removeHeader("Authorization");
+    servletRequest.addHeader("Authorization", "value2");
+    assertEquals("value2", servletRequest.getHeader("Authorization"));
+    final String jobId2 = assertExecutedAsync();
+    assertNotEquals(jobId1, jobId2);
+  }
+
+  @Test
+  public void testCreatesNewAsyncJobWhenDatabaseVersionChanges() {
+    when(database.getCacheKey()).thenReturn(Optional.of("key1"));
+    final String jobId1 = assertExecutedAsync();
+    when(database.getCacheKey()).thenReturn(Optional.of("key2"));
+    final String jobId2 = assertExecutedAsync();
+    assertNotEquals(jobId1, jobId2);
+  }
+
+  @Test
+  public void testCreatesNewAsyncJobWhenQueryStringChanges() {
+    requestDetails.setCompleteUrl(FHIR_SERVER_BASE + "/Patient/$aggregate?param=value1");
+    final String jobId1 = assertExecutedAsync();
+    requestDetails.setCompleteUrl(FHIR_SERVER_BASE + "/Patient/$aggregate?param=value2");
+    final String jobId2 = assertExecutedAsync();
+    assertNotEquals(jobId1, jobId2);
+  }
+
+  @Test
+  public void testReusesAsyncJobWhenAuthenticationPrincipalChanges() {
+    setAuthenticationPrincipal("principal1");
+    final String jobId1 = assertExecutedAsync();
+    setAuthenticationPrincipal("principal2");
+    final String jobId2 = assertExecutedAsync();
+    assertEquals(jobId1, jobId2);
+  }
+}
diff --git a/fhir-server/src/test/java/au/csiro/pathling/async/JobRegistryTest.java b/fhir-server/src/test/java/au/csiro/pathling/async/JobRegistryTest.java
new file mode 100644
index 0000000000..23782a327f
--- /dev/null
+++ b/fhir-server/src/test/java/au/csiro/pathling/async/JobRegistryTest.java
@@ -0,0 +1,77 @@
+/*
+ * Copyright 2023 Commonwealth Scientific and Industrial Research
+ * Organisation (CSIRO) ABN 41 687 119 230.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package au.csiro.pathling.async;
+
+import au.csiro.pathling.async.Job.JobTag;
+import org.hl7.fhir.instance.model.api.IBaseResource;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import javax.annotation.Nonnull;
+import java.util.Optional;
+import java.util.concurrent.Future;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.Mockito.mock;
+
+class JobRegistryTest {
+
+  private final static Future<IBaseResource> MOCK_FUTURE = mock(Future.class);
+  private final static JobTag JOB_TAG_1 = new JobTag() {
+  };
+  private final static JobTag JOB_TAG_2 = new JobTag() {
+  };
+
+  @Nonnull
+  private final JobRegistry registry = new JobRegistry();
+
+  @Test
+  void testNewJobCanBeRetrievedById() {
+    final Job newJob = registry.getOrCreate(JOB_TAG_1,
+        id -> new Job(id, "operation", MOCK_FUTURE, Optional.empty()));
+
+    assertEquals(newJob, registry.get(newJob.getId()));
+  }
+
+  @Test
+  void testReusesJobWhenTagsAreIdentical() {
+    assertEquals(JOB_TAG_1, JOB_TAG_1);
+    final Job firstJob = registry.getOrCreate(JOB_TAG_1,
+        id -> new Job(id, "operation", MOCK_FUTURE, Optional.empty()));
+    final Job otherJob = registry.getOrCreate(JOB_TAG_1,
+        id -> new Job(id, "operation", MOCK_FUTURE, Optional.empty()));
+
+    assertEquals(firstJob, otherJob);
+    assertEquals(firstJob, registry.get(firstJob.getId()));
+  }
+
+  @Test
+  void testCreatesNewJobIfTagsDiffer() {
+
+    assertNotEquals(JOB_TAG_1, JOB_TAG_2);
+
+    final Job firstJob = registry.getOrCreate(JOB_TAG_1,
+        id -> new Job(id, "operation", MOCK_FUTURE, Optional.empty()));
+
+    final Job otherJob = registry.getOrCreate(JOB_TAG_2,
+        id -> new Job(id, "operation", MOCK_FUTURE, Optional.empty()));
+
+    assertNotEquals(firstJob, otherJob);
+    assertNotEquals(firstJob.getId(), otherJob.getId());
+  }
+}
diff --git a/fhir-server/src/test/java/au/csiro/pathling/async/RequestTagFactoryTest.java b/fhir-server/src/test/java/au/csiro/pathling/async/RequestTagFactoryTest.java
new file mode 100644
index 0000000000..e72b43ca0c
--- /dev/null
+++ b/fhir-server/src/test/java/au/csiro/pathling/async/RequestTagFactoryTest.java
@@ -0,0 +1,141 @@
+/*
+ * Copyright 2023 Commonwealth Scientific and Industrial Research
+ * Organisation (CSIRO) ABN 41 687 119 230.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package au.csiro.pathling.async;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import au.csiro.pathling.caching.Cacheable;
+import au.csiro.pathling.config.AsyncConfiguration;
+import au.csiro.pathling.config.HttpServerCachingConfiguration;
+import au.csiro.pathling.config.ServerConfiguration;
+import au.csiro.pathling.io.CacheableDatabase;
+import ca.uhn.fhir.rest.server.servlet.ServletRequestDetails;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.junit.jupiter.api.Test;
+import org.springframework.security.core.Authentication;
+
+public class RequestTagFactoryTest {
+
+  private final CacheableDatabase mockDatabase = mock(CacheableDatabase.class);
+  private final Cacheable mockCacheable = mock(Cacheable.class);
+  private final ServletRequestDetails mockRequestDetails = mock(ServletRequestDetails.class);
+  private final Authentication mockAuthentication = mock(Authentication.class);
+
+  @Test
+  public void testComputesSalientHeaderNamesCorrectly() {
+
+    final List<String> varyHeaders = List.of("Accept", "Accept-Encoding", "Authorization");
+    final List<String> whiteListedHeaders = List.of("Accept", "Accept-Encoding", "X-UserId");
+
+    assertEquals(Collections.emptySet(), new RequestTagFactory(
+        mockDatabase,
+        createServerConfiguration(Collections.emptyList(),
+            Collections.emptyList())).getSalientHeaderNames());
+
+    assertEquals(Collections.emptySet(), new RequestTagFactory(
+        mockDatabase,
+        createServerConfiguration(Collections.emptyList(),
+            whiteListedHeaders)).getSalientHeaderNames());
+
+    assertEquals(
+        varyHeaders.stream().map(String::toLowerCase).collect(Collectors.toUnmodifiableSet()),
+        new RequestTagFactory(
+            mockDatabase,
+            createServerConfiguration(varyHeaders,
+                Collections.emptyList())).getSalientHeaderNames());
+
+    assertEquals(Set.of("authorization"), new RequestTagFactory(
+        mockDatabase,
+        createServerConfiguration(varyHeaders,
+            whiteListedHeaders)).getSalientHeaderNames());
+  }
+
+  @Test
+  public void testComputesCorrectTagForAuthenticatedUserAndExisingCacheKey() {
+    final Set<String> salientHeaderNames = Set.of("X-Single-Value", "X-Multi-Values",
+        "X-Not-Present");
+    // RequestDetails.gethHeaders() returns a Map<String, List<String>> where the keys are
+    // lower-cased.
+    final Map<String, List<String>> requestHeaders = Map.of("x-single-value",
+        List.of("singleValue"),
+        "x-multi-values", List.of("multiValue1", "multiValue2"), "x-other", List.of("otherValue"));
+    final Object principal = new Object();
+    when(mockAuthentication.getPrincipal()).thenReturn(principal);
+
+    when(mockCacheable.getCacheKey()).thenReturn(Optional.of("cacheKey_A"));
+    when(mockRequestDetails.getCompleteUrl()).thenReturn("uri:requestUri-A");
+    when(mockRequestDetails.getHeaders()).thenReturn(requestHeaders);
+
+    final RequestTagFactory requestTagFactory = new RequestTagFactory(mockCacheable,
+        salientHeaderNames);
+    final RequestTag requestTag = requestTagFactory.createTag(mockRequestDetails,
+        mockAuthentication);
+    assertEquals(new RequestTag("uri:requestUri-A",
+            Map.of("x-single-value", List.of("singleValue"),
+                "x-multi-values", List.of("multiValue1", "multiValue2")),
+            Optional.of("cacheKey_A")),
+        requestTag);
+  }
+
+  @Test
+  public void testComputesCorrectTagFoNoPrincipalAndMissingKey() {
+    final Set<String> salientHeaderNames = Set.of("Y-SingleValue", "Y-MultiValues", "Y-NotPresent");
+    final Map<String, List<String>> requestHeaders = Map.of("Y-SingleValue", List.of("singleValue"),
+        "Y-MultiValues", List.of("multiValue1", "multiValue2"), "Y-Other", List.of("otherValue"));
+
+    when(mockCacheable.getCacheKey()).thenReturn(Optional.empty());
+    when(mockRequestDetails.getCompleteUrl()).thenReturn("uri:requestUri-B");
+    when(mockRequestDetails.getHeaders()).thenReturn(requestHeaders);
+
+    final RequestTagFactory requestTagFactory = new RequestTagFactory(mockDatabase,
+        createServerConfiguration(List.of("X-Single-Value", "X-Multi-Values",
+            "X-Not-Present"), Collections.emptyList()));
+
+    final RequestTag requestTag = requestTagFactory.createTag(mockRequestDetails,
+        mockAuthentication);
+    assertEquals(new RequestTag("uri:requestUri-B",
+            Map.of("Y-SingleValue", List.of("singleValue"),
+                "Y-MultiValues", List.of("multiValue1", "multiValue2")),
+            Optional.empty()),
+        requestTag);
+  }
+
+
+  static ServerConfiguration createServerConfiguration(final List<String> varyHeaders,
+      final List<String> excludeVary) {
+    final AsyncConfiguration asyncConfiguration = new AsyncConfiguration();
+    asyncConfiguration.setEnabled(true);
+    asyncConfiguration.setVaryHeadersExcludedFromCacheKey(excludeVary);
+
+    final HttpServerCachingConfiguration httpServerCachingConfiguration = new HttpServerCachingConfiguration();
+    httpServerCachingConfiguration.setVary(varyHeaders);
+
+    final ServerConfiguration serverConfiguration = new ServerConfiguration();
+    serverConfiguration.setAsync(asyncConfiguration);
+    serverConfiguration.setHttpCaching(httpServerCachingConfiguration);
+    return serverConfiguration;
+  }
+
+}
diff --git a/fhir-server/src/test/java/au/csiro/pathling/security/SecurityTestForOperations.java b/fhir-server/src/test/java/au/csiro/pathling/security/SecurityTestForOperations.java
index 5ec3afa0f8..4021721e36 100644
--- a/fhir-server/src/test/java/au/csiro/pathling/security/SecurityTestForOperations.java
+++ b/fhir-server/src/test/java/au/csiro/pathling/security/SecurityTestForOperations.java
@@ -72,7 +72,7 @@ void setUp() {
 
   void assertImportSuccess() {
     try {
-      importProvider.importOperation(new Parameters(), null, null, null);
+      importProvider.importOperation(new Parameters(), null);
     } catch (final InvalidUserInputError ex) {
       // pass
     }
@@ -82,7 +82,7 @@ void assertAggregateSuccess() {
     final AggregateProvider aggregateProvider = (AggregateProvider) resourceProviderFactory
         .createAggregateResourceProvider(ResourceType.Patient);
     try {
-      aggregateProvider.aggregate(null, null, null, null, null, null);
+      aggregateProvider.aggregate(null, null, null, null);
     } catch (final InvalidUserInputError ex) {
       // pass
     }
diff --git a/fhir-server/src/test/java/au/csiro/pathling/test/integration/AsyncTest.java b/fhir-server/src/test/java/au/csiro/pathling/test/integration/AsyncTest.java
index 0045f24ec4..88ff49b59f 100644
--- a/fhir-server/src/test/java/au/csiro/pathling/test/integration/AsyncTest.java
+++ b/fhir-server/src/test/java/au/csiro/pathling/test/integration/AsyncTest.java
@@ -125,10 +125,7 @@ void assertAsyncResponse(@Nonnull final ResponseEntity<String> response,
         throw new AssertionError("Timed out waiting for async result");
       }
 
-      final List<String> contentLocations = response.getHeaders().get("Content-Location");
-      assertNotNull(contentLocations);
-      final String contentLocation = contentLocations.get(0);
-      assertNotNull(contentLocation);
+      final String contentLocation = getContentLocation(response);
       final URL statusUrl = new URL(contentLocation);
       final RequestEntity<Void> statusRequest = RequestEntity.get(statusUrl.toURI()).build();
       log.info("Sending status request");
@@ -160,4 +157,36 @@ void assertAsyncResponse(@Nonnull final ResponseEntity<String> response,
     log.info("Successful response received");
   }
 
+  @Nonnull
+  private String getContentLocation(@Nonnull ResponseEntity<String> response) {
+    final List<String> contentLocations = response.getHeaders().get("Content-Location");
+    assertNotNull(contentLocations);
+    final String contentLocation = contentLocations.get(0);
+    assertNotNull(contentLocation);
+    return contentLocation;
+  }
+
+
+  @Test
+  void identicalAsyncRequestsReturnTheSameJobId()
+      throws URISyntaxException {
+    TestHelpers.mockResource(database, spark, ResourceType.OBSERVATION);
+    final String uri = "http://localhost:" + port + "/fhir/Observation/$extract?column=id&"
+        + "column=code.coding&column=valueQuantity.value&column=valueQuantity.unit";
+    final RequestEntity<Void> request = RequestEntity.get(new URI(uri))
+        .header("Prefer", "respond-async").build();
+    log.info("Sending kick-off request");
+    final ResponseEntity<String> response1 =
+        restTemplate.exchange(uri, HttpMethod.GET, request, String.class);
+    assertEquals(HttpStatus.ACCEPTED_202, response1.getStatusCode().value());
+
+    final ResponseEntity<String> response2 =
+        restTemplate.exchange(uri, HttpMethod.GET, request, String.class);
+    assertEquals(HttpStatus.ACCEPTED_202, response1.getStatusCode().value());
+    assertEquals(getContentLocation(response1), getContentLocation(response2));
+  }
+
+  
+  
+  
 }