Merge branch 'main' into feat/2258-edit-group-info

.github/workflows/ena-submission-image.yaml

@@ -36,6 +36,19 @@ jobs:
       packages: write
       checks: read
     steps:
+      - name: Checkout repository
+        uses: actions/checkout@v4
+      - name: Validate submitToEnaProduction is not true in values.yaml
+        run: |
+          python -c "
+          import yaml
+          with open('kubernetes/loculus/values.yaml', 'r') as file:
+            values = yaml.safe_load(file)
+          submit_to_ena_prod = values.get('submitToEnaProduction', False)
+          if submit_to_ena_prod:
+            print('Error: The flag submitToEnaProduction is set to true - this will submit data to ENA production. Please set it to false in values.yaml')
+            exit(1)
+          "
       - name: Shorten sha
         run: echo "sha=${sha::7}" >> $GITHUB_ENV
       - uses: actions/checkout@v4

.github/workflows/update-argocd-metadata.yml

@@ -74,6 +74,13 @@ jobs:
           check-name: Build keycloakify Docker Image
           repo-token: ${{ secrets.GITHUB_TOKEN }}
           wait-interval: 2
+      - name: Wait for ENA Submission Docker Image
+        uses: lewagon/[email protected]
+        with:
+          ref: ${{ github.sha }}
+          check-name: Build ena-submission Docker Image
+          repo-token: ${{ secrets.GITHUB_TOKEN }}
+          wait-interval: 2
       # End of wait block
       - name: Checkout External Repository
         uses: actions/checkout@v4

README.md

@@ -60,6 +60,7 @@ While the documentation is still a work in progress, a look at the [`.github/wor
 
 ## Authorization
 
+
 ### User management
 
 We use keycloak for authorization. The keycloak instance is deployed in the `loculus` namespace and exposed to the outside either under `localhost:8083` or `authentication-[your-argo-cd-path]`. The keycloak instance is configured with a realm called `loculus` and a client called `backend-client`. The realm is configured to use the exposed url of keycloak as a [frontend url](https://www.keycloak.org/server/hostname).

backend/build.gradle

@@ -50,7 +50,7 @@ dependencies {
     implementation "org.springframework.boot:spring-boot-starter-security"
 
     implementation 'org.apache.commons:commons-compress:1.27.1'
-    implementation 'com.github.luben:zstd-jni:1.5.6-5'
+    implementation 'com.github.luben:zstd-jni:1.5.6-6'
     implementation 'org.tukaani:xz:1.10'
 
     implementation("org.redundent:kotlin-xml-builder:1.9.1")

backend/src/main/kotlin/org/loculus/backend/controller/SubmissionController.kt

@@ -133,6 +133,19 @@ open class SubmissionController(
                 schema = Schema(implementation = UnprocessedData::class),
             ),
         ],
+        headers = [
+            Header(
+                name = "eTag",
+                description = "Last database write Etag",
+                schema = Schema(type = "integer"),
+            ),
+        ],
+    )
+    @ApiResponse(
+        responseCode = "304",
+        description =
+        "No database changes since last request " +
+            "(Etag in HttpHeaders.IF_NONE_MATCH matches lastDatabaseWriteETag)",
     )
     @ApiResponse(responseCode = "422", description = EXTRACT_UNPROCESSED_DATA_ERROR_RESPONSE)
     @PostMapping("/extract-unprocessed-data", produces = [MediaType.APPLICATION_NDJSON_VALUE])
@@ -143,6 +156,7 @@ open class SubmissionController(
             message = "You can extract at max $MAX_EXTRACTED_SEQUENCE_ENTRIES sequence entries at once.",
         ) numberOfSequenceEntries: Int,
         @RequestParam pipelineVersion: Long,
+        @RequestHeader(value = HttpHeaders.IF_NONE_MATCH, required = false) ifNoneMatch: String?,
     ): ResponseEntity<StreamingResponseBody> {
         val currentProcessingPipelineVersion = submissionDatabaseService.getCurrentProcessingPipelineVersion()
         if (pipelineVersion < currentProcessingPipelineVersion) {
@@ -152,8 +166,12 @@ open class SubmissionController(
             )
         }
 
+        val lastDatabaseWriteETag = releasedDataModel.getLastDatabaseWriteETag()
+        if (ifNoneMatch == lastDatabaseWriteETag) return ResponseEntity.status(HttpStatus.NOT_MODIFIED).build()
+
         val headers = HttpHeaders()
         headers.contentType = MediaType.parseMediaType(MediaType.APPLICATION_NDJSON_VALUE)
+        headers.eTag = lastDatabaseWriteETag
         val streamBody = streamTransactioned {
             submissionDatabaseService.streamUnprocessedSubmissions(numberOfSequenceEntries, organism, pipelineVersion)
         }
@@ -349,10 +367,17 @@ open class SubmissionController(
     @ApiResponse(
         responseCode = "200",
         description = GET_ORIGINAL_METADATA_RESPONSE_DESCRIPTION,
+        headers = [
+            Header(
+                name = "x-total-records",
+                description = "The total number of records sent in responseBody",
+                schema = Schema(type = "integer"),
+            ),
+        ],
     )
     @ApiResponse(
         responseCode = "423",
-        description = "Locked. The metadata is currently being processed.",
+        description = "Locked. New sequence entries are currently being uploaded.",
     )
     @GetMapping("/get-original-metadata", produces = [MediaType.APPLICATION_JSON_VALUE])
     fun getOriginalMetadata(
@@ -369,16 +394,29 @@ open class SubmissionController(
         @HiddenParam authenticatedUser: AuthenticatedUser,
         @RequestParam compression: CompressionFormat?,
     ): ResponseEntity<StreamingResponseBody> {
+        val stillProcessing = submitModel.checkIfStillProcessingSubmittedData()
+        if (stillProcessing) {
+            return ResponseEntity.status(HttpStatus.LOCKED).build()
+        }
+
         val headers = HttpHeaders()
         headers.contentType = MediaType.parseMediaType(MediaType.APPLICATION_NDJSON_VALUE)
         if (compression != null) {
             headers.add(HttpHeaders.CONTENT_ENCODING, compression.compressionName)
         }
 
-        val stillProcessing = submissionDatabaseService.checkIfStillProcessingSubmittedData()
-        if (stillProcessing) {
-            return ResponseEntity.status(HttpStatus.LOCKED).build()
-        }
+        val totalRecords = submissionDatabaseService.countOriginalMetadata(
+            authenticatedUser,
+            organism,
+            groupIdsFilter?.takeIf { it.isNotEmpty() },
+            statusesFilter?.takeIf { it.isNotEmpty() },
+        )
+        headers.add("x-total-records", totalRecords.toString())
+        // TODO(https://github.com/loculus-project/loculus/issues/2778)
+        // There's a possibility that the totalRecords change between the count and the actual query
+        // this is not too bad, if the client ends up with a few more records than expected
+        // We just need to make sure the etag used is from before the count
+        // Alternatively, we could read once to file while counting and then stream the file
 
         val streamBody = streamTransactioned(compression) {
             submissionDatabaseService.streamOriginalMetadata(

backend/src/main/kotlin/org/loculus/backend/controller/SubmissionControllerDescriptions.kt

@@ -146,6 +146,8 @@ and roll back the whole transaction.
 const val GET_RELEASED_DATA_DESCRIPTION = """
 Get released data as a stream of NDJSON.
 This returns all accession versions that have the status 'APPROVED_FOR_RELEASE'. 
+Optionally submit the etag received in previous request with If-None-Match
+to only retrieve all released data if the database has changed since last request.
 """
 
 const val GET_RELEASED_DATA_RESPONSE_DESCRIPTION = """

backend/src/main/kotlin/org/loculus/backend/model/SubmitModel.kt

@@ -17,11 +17,14 @@ import org.loculus.backend.controller.UnprocessableEntityException
 import org.loculus.backend.service.datauseterms.DataUseTermsPreconditionValidator
 import org.loculus.backend.service.groupmanagement.GroupManagementPreconditionValidator
 import org.loculus.backend.service.submission.CompressionAlgorithm
+import org.loculus.backend.service.submission.MetadataUploadAuxTable
+import org.loculus.backend.service.submission.SequenceUploadAuxTable
 import org.loculus.backend.service.submission.UploadDatabaseService
 import org.loculus.backend.utils.FastaReader
 import org.loculus.backend.utils.metadataEntryStreamAsSequence
 import org.loculus.backend.utils.revisionEntryStreamAsSequence
 import org.springframework.stereotype.Service
+import org.springframework.transaction.annotation.Transactional
 import org.springframework.web.multipart.MultipartFile
 import java.io.BufferedInputStream
 import java.io.File
@@ -313,4 +316,13 @@ class SubmitModel(
             throw UnprocessableEntityException(metadataNotPresentErrorText + sequenceNotPresentErrorText)
         }
     }
+
+    @Transactional(readOnly = true)
+    fun checkIfStillProcessingSubmittedData(): Boolean {
+        val metadataInAuxTable: Boolean =
+            MetadataUploadAuxTable.select(MetadataUploadAuxTable.submissionIdColumn).count() > 0
+        val sequencesInAuxTable: Boolean =
+            SequenceUploadAuxTable.select(SequenceUploadAuxTable.sequenceSubmissionIdColumn).count() > 0
+        return metadataInAuxTable || sequencesInAuxTable
+    }
 }

backend/src/main/kotlin/org/loculus/backend/service/submission/SubmissionDatabaseService.kt

@@ -950,21 +950,12 @@ 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(
+    private fun originalMetadataFilter(
         authenticatedUser: AuthenticatedUser,
         organism: Organism,
         groupIdsFilter: List<Int>?,
         statusesFilter: List<Status>?,
-        fields: List<String>?,
-    ): Sequence<AccessionVersionOriginalMetadata> {
+    ): Op<Boolean> {
         val organismCondition = SequenceEntriesView.organismIs(organism)
         val groupCondition = getGroupCondition(groupIdsFilter, authenticatedUser)
         val statusCondition = if (statusesFilter != null) {
@@ -974,6 +965,33 @@ open class SubmissionDatabaseService(
         }
         val conditions = organismCondition and groupCondition and statusCondition
 
+        return conditions
+    }
+
+    fun countOriginalMetadata(
+        authenticatedUser: AuthenticatedUser,
+        organism: Organism,
+        groupIdsFilter: List<Int>?,
+        statusesFilter: List<Status>?,
+    ): Long = SequenceEntriesView
+        .selectAll()
+        .where(
+            originalMetadataFilter(
+                authenticatedUser,
+                organism,
+                groupIdsFilter,
+                statusesFilter,
+            ),
+        )
+        .count()
+
+    fun streamOriginalMetadata(
+        authenticatedUser: AuthenticatedUser,
+        organism: Organism,
+        groupIdsFilter: List<Int>?,
+        statusesFilter: List<Status>?,
+        fields: List<String>?,
+    ): Sequence<AccessionVersionOriginalMetadata> {
         val originalMetadata = SequenceEntriesView.originalDataColumn
             .extract<Map<String, String>>("metadata")
             .alias("original_metadata")
@@ -984,7 +1002,14 @@ open class SubmissionDatabaseService(
                 SequenceEntriesView.accessionColumn,
                 SequenceEntriesView.versionColumn,
             )
-            .where(conditions)
+            .where(
+                originalMetadataFilter(
+                    authenticatedUser,
+                    organism,
+                    groupIdsFilter,
+                    statusesFilter,
+                ),
+            )
             .fetchSize(streamBatchSize)
             .asSequence()
             .map {

backend/src/main/kotlin/org/loculus/backend/service/submission/UploadDatabaseService.kt

@@ -7,7 +7,6 @@ import org.jetbrains.exposed.sql.VarCharColumnType
 import org.jetbrains.exposed.sql.and
 import org.jetbrains.exposed.sql.batchInsert
 import org.jetbrains.exposed.sql.deleteWhere
-import org.jetbrains.exposed.sql.select
 import org.jetbrains.exposed.sql.selectAll
 import org.jetbrains.exposed.sql.statements.StatementType
 import org.jetbrains.exposed.sql.transactions.transaction

backend/src/test/kotlin/org/loculus/backend/controller/submission/ExtractUnprocessedDataEndpointTest.kt

@@ -27,7 +27,10 @@ import org.loculus.backend.controller.expectUnauthorizedResponse
 import org.loculus.backend.controller.getAccessionVersions
 import org.loculus.backend.controller.jwtForDefaultUser
 import org.loculus.backend.controller.submission.SubmitFiles.DefaultFiles
+import org.loculus.backend.controller.submission.SubmitFiles.DefaultFiles.NUMBER_OF_SEQUENCES
 import org.springframework.beans.factory.annotation.Autowired
+import org.springframework.http.HttpHeaders.ETAG
+import org.springframework.test.web.servlet.result.MockMvcResultMatchers.header
 import org.springframework.test.web.servlet.result.MockMvcResultMatchers.jsonPath
 import org.springframework.test.web.servlet.result.MockMvcResultMatchers.status
 
@@ -67,6 +70,35 @@ class ExtractUnprocessedDataEndpointTest(
         assertThat(responseBody, `is`(emptyList()))
     }
 
+    @Test
+    fun `GIVEN header etag equal etag from last db update THEN respond with 304, ELSE respond with data and etag`() {
+        val submissionResult = convenienceClient.submitDefaultFiles()
+        val response = client.extractUnprocessedData(DefaultFiles.NUMBER_OF_SEQUENCES)
+
+        val initialEtag = response.andReturn().response.getHeader(ETAG)
+
+        val responseBody = response.expectNdjsonAndGetContent<UnprocessedData>()
+        assertThat(responseBody.size, `is`(DefaultFiles.NUMBER_OF_SEQUENCES))
+
+        val responseAfterUpdatingTable = client.extractUnprocessedData(
+            DefaultFiles.NUMBER_OF_SEQUENCES,
+            ifNoneMatch = initialEtag,
+        ).andExpect(status().isOk)
+
+        val emptyResponseBody = responseAfterUpdatingTable.expectNdjsonAndGetContent<UnprocessedData>()
+        assertThat(emptyResponseBody.size, `is`(0))
+
+        val secondEtag = responseAfterUpdatingTable.andReturn().response.getHeader(ETAG)
+
+        val responseNoNewData = client.extractUnprocessedData(
+            DefaultFiles.NUMBER_OF_SEQUENCES,
+            ifNoneMatch = secondEtag,
+        )
+
+        responseNoNewData.andExpect(status().isNotModified)
+            .andExpect(header().doesNotExist(ETAG))
+    }
+
     @Test
     fun `WHEN extracting unprocessed data THEN only previously not extracted sequence entries are returned`() {
         val submissionResult = convenienceClient.submitDefaultFiles()