feat(backend, ingest): Only return get-original-data when the aux-tab…

…le is empty (#2846) *get-original-metadata responds with service LOCKED if there is data in aux table. * Make get-submitted retry if loculus returns 423 * Update ingest get-submitted to not add "" to dictionary when segment is missing and thus insdc_accession_segment does not exist. --------- Co-authored-by: Cornelius Roemer <cornelius.roemer@gmail.com>
loculus-project · Sep 19, 2024 · 61b8fde · 61b8fde
1 parent aff8472
commit 61b8fde
Show file tree

Hide file tree

Showing 3 changed files with 32 additions and 7 deletions.
diff --git a/backend/src/main/kotlin/org/loculus/backend/controller/SubmissionController.kt b/backend/src/main/kotlin/org/loculus/backend/controller/SubmissionController.kt
@@ -77,8 +77,7 @@ open class SubmissionController(
     @ApiResponse(responseCode = "200", description = SUBMIT_RESPONSE_DESCRIPTION)
     @PostMapping("/submit", consumes = ["multipart/form-data"])
     fun submit(
-        @PathVariable
-        @Valid organism: Organism,
+        @PathVariable @Valid organism: Organism,
         @HiddenParam authenticatedUser: AuthenticatedUser,
         @Parameter(description = GROUP_ID_DESCRIPTION) @RequestParam groupId: Int,
         @Parameter(description = METADATA_FILE_DESCRIPTION) @RequestParam metadataFile: MultipartFile,
@@ -255,15 +254,17 @@ open class SubmissionController(
     )
     @ApiResponse(
         responseCode = "304",
-        description = "No database changes since last request " +
+        description =
+        "No database changes since last request " +
             "(Etag in HttpHeaders.IF_NONE_MATCH matches lastDatabaseWriteETag)",
     )
     @GetMapping("/get-released-data", produces = [MediaType.APPLICATION_NDJSON_VALUE])
     fun getReleasedData(
         @PathVariable @Valid organism: Organism,
         @RequestParam compression: CompressionFormat?,
-        @Parameter(description = "(Optional) Only retrieve all released data if Etag has changed.")
-        @RequestHeader(value = HttpHeaders.IF_NONE_MATCH, required = false) ifNoneMatch: String?,
+        @Parameter(
+            description = "(Optional) Only retrieve all released data if Etag has changed.",
+        ) @RequestHeader(value = HttpHeaders.IF_NONE_MATCH, required = false) ifNoneMatch: String?,
     ): ResponseEntity<StreamingResponseBody> {
         val lastDatabaseWriteETag = releasedDataModel.getLastDatabaseWriteETag(
             RELEASED_DATA_RELATED_TABLES,
@@ -349,6 +350,10 @@ open class SubmissionController(
         responseCode = "200",
         description = GET_ORIGINAL_METADATA_RESPONSE_DESCRIPTION,
     )
+    @ApiResponse(
+        responseCode = "423",
+        description = "Locked. The metadata is currently being processed.",
+    )
     @GetMapping("/get-original-metadata", produces = [MediaType.APPLICATION_JSON_VALUE])
     fun getOriginalMetadata(
         @PathVariable @Valid organism: Organism,
@@ -370,6 +375,11 @@ open class SubmissionController(
             headers.add(HttpHeaders.CONTENT_ENCODING, compression.compressionName)
         }
 
+        val stillProcessing = submissionDatabaseService.checkIfStillProcessingSubmittedData()
+        if (stillProcessing) {
+            return ResponseEntity.status(HttpStatus.LOCKED).build()
+        }
+
         val streamBody = streamTransactioned(compression) {
             submissionDatabaseService.streamOriginalMetadata(
                 authenticatedUser,

diff --git a/backend/src/main/kotlin/org/loculus/backend/service/submission/SubmissionDatabaseService.kt b/backend/src/main/kotlin/org/loculus/backend/service/submission/SubmissionDatabaseService.kt
@@ -561,7 +561,7 @@ open class SubmissionDatabaseService(
         )
     }.count()
 
-// Make sure to keep in sync with countReleasedSubmissions query
+    // Make sure to keep in sync with countReleasedSubmissions query
     fun streamReleasedSubmissions(organism: Organism): Sequence<RawProcessedData> = SequenceEntriesView.join(
         DataUseTermsTable,
         JoinType.LEFT,
@@ -950,6 +950,14 @@ open class SubmissionDatabaseService(
         )
     }
 
+    fun checkIfStillProcessingSubmittedData(): Boolean {
+        val metadataInAuxTable: Boolean =
+            MetadataUploadAuxTable.select(MetadataUploadAuxTable.submissionIdColumn).count() > 0
+        val sequencesInAuxTable: Boolean =
+            SequenceUploadAuxTable.select(SequenceUploadAuxTable.sequenceSubmissionIdColumn).count() > 0
+        return metadataInAuxTable || sequencesInAuxTable
+    }
+
     fun streamOriginalMetadata(
         authenticatedUser: AuthenticatedUser,
         organism: Organism,

diff --git a/ingest/scripts/call_loculus.py b/ingest/scripts/call_loculus.py
@@ -105,6 +105,11 @@ def make_request(  # noqa: PLR0913, PLR0917
             msg = f"Unsupported HTTP method: {method}"
             raise ValueError(msg)
 
+    if response.status_code == 423:
+        logger.warning(f"Got 423 from {url}. Retrying after 30 seconds.")
+        sleep(30)
+        return make_request(method, url, config, params, files, json_body)
+
     if not response.ok:
         error_message = (
             f"Request failed:\n"
@@ -338,7 +343,9 @@ def get_submitted(config: Config):
         original_metadata: dict[str, str] = entry["originalMetadata"]
         hash_value = original_metadata.get("hash", "")
         if config.segmented:
-            insdc_accessions = [original_metadata[key] for key in insdc_key]
+            insdc_accessions = [
+                original_metadata[key] for key in insdc_key if original_metadata[key]
+            ]
             joint_accession = "/".join(
                 [
                     f"{original_metadata[key]}.{segment}"