From bd23e8e08801660f9b5551bcb5fb45b74715a45a Mon Sep 17 00:00:00 2001 From: stacy Date: Mon, 20 Jul 2026 09:50:51 +0200 Subject: [PATCH] fix(backend): spool released exports safely Build each export from one successful query result so X-Total-Records matches the served bytes and the database transaction ends before transfer. Write security headers before the async file copy. Bound spool storage and clean it across retries, failed handoffs, timeouts, disconnects, and restarts. Align the bundled download clients and deployment timeouts with generating the full spool before the first byte is sent. Refs #3511, #5198 --- .../backend/config/BackendSpringConfig.kt | 6 + .../ReleasedDataSpoolCleanupInterceptor.kt | 47 +++ .../loculus/backend/config/SecurityConfig.kt | 11 + .../org/loculus/backend/config/WebConfig.kt | 8 + .../backend/controller/ExceptionHandler.kt | 16 +- .../controller/SubmissionController.kt | 65 +++- .../service/submission/StreamSpoolService.kt | 300 ++++++++++++++++++ .../submission/SubmissionDatabaseService.kt | 15 - .../loculus/backend/utils/IteratorStreamer.kt | 14 +- .../src/main/resources/application.properties | 11 + ...ReleasedDataSpoolCleanupInterceptorTest.kt | 91 ++++++ .../loculus/backend/controller/TestHelpers.kt | 5 +- .../submission/GetReleasedDataEndpointTest.kt | 31 ++ .../GetReleasedDataSpoolEndpointTest.kt | 160 ++++++++++ .../submission/StreamSpoolServiceTest.kt | 98 ++++++ .../backend/utils/IteratorStreamerTest.kt | 45 +++ cli/src/loculus_cli/api/backend.py | 15 +- cli/tests/test_backend.py | 27 ++ .../setup-with-k3d-and-nginx.mdx | 1 + .../templates/ena-submission-deployment.yaml | 2 +- .../loculus/templates/loculus-backend.yaml | 11 + .../loculus/templates/silo-deployment.yaml | 6 +- kubernetes/loculus/values.schema.json | 87 ++++- kubernetes/loculus/values.yaml | 15 +- loculus-silo/README.md | 1 + loculus-silo/src/silo_import/config.py | 4 + .../src/silo_import/download_manager.py | 14 +- loculus-silo/tests/test_config.py | 15 + loculus-silo/tests/test_download_manager.py | 81 +++++ loculus-silo/tests/test_integration.py | 1 + loculus-silo/tests/test_runner.py | 1 + 31 files changed, 1152 insertions(+), 52 deletions(-) create mode 100644 backend/src/main/kotlin/org/loculus/backend/config/ReleasedDataSpoolCleanupInterceptor.kt create mode 100644 backend/src/main/kotlin/org/loculus/backend/service/submission/StreamSpoolService.kt create mode 100644 backend/src/test/kotlin/org/loculus/backend/config/ReleasedDataSpoolCleanupInterceptorTest.kt create mode 100644 backend/src/test/kotlin/org/loculus/backend/controller/submission/GetReleasedDataSpoolEndpointTest.kt create mode 100644 backend/src/test/kotlin/org/loculus/backend/service/submission/StreamSpoolServiceTest.kt create mode 100644 backend/src/test/kotlin/org/loculus/backend/utils/IteratorStreamerTest.kt create mode 100644 cli/tests/test_backend.py create mode 100644 loculus-silo/tests/test_download_manager.py diff --git a/backend/src/main/kotlin/org/loculus/backend/config/BackendSpringConfig.kt b/backend/src/main/kotlin/org/loculus/backend/config/BackendSpringConfig.kt index 89f2f48a5f..2b7cbf22fa 100644 --- a/backend/src/main/kotlin/org/loculus/backend/config/BackendSpringConfig.kt +++ b/backend/src/main/kotlin/org/loculus/backend/config/BackendSpringConfig.kt @@ -39,6 +39,12 @@ object BackendSpringProperty { const val PIPELINE_VERSION_UPGRADE_CHECK_INTERVAL_SECONDS = "loculus.pipeline-version-upgrade-check.interval-seconds" const val STREAM_BATCH_SIZE = "loculus.stream.batch-size" + const val STREAM_SPOOL_DIR = "loculus.stream.spool-dir" + const val STREAM_MAX_CONCURRENT_SPOOLS = "loculus.stream.max-concurrent-spools" + const val STREAM_SPOOL_MAX_TOTAL_BYTES = "loculus.stream.spool-max-total-bytes" + const val STREAM_SPOOL_MIN_FREE_BYTES = "loculus.stream.spool-min-free-bytes" + const val STREAM_SPOOL_FILE_TTL_MINUTES = "loculus.stream.spool-file-ttl-minutes" + const val STREAM_SPOOL_SWEEP_EVERY_MINUTES = "loculus.stream.spool-sweep-every-minutes" const val DEBUG_MODE = "loculus.debug-mode" const val ENABLE_SEQSETS = "loculus.enable-seqsets" const val SEQSET_CITATIONS_RUN_EVERY_MINUTES = "loculus.seqset-citations.run-every-minutes" diff --git a/backend/src/main/kotlin/org/loculus/backend/config/ReleasedDataSpoolCleanupInterceptor.kt b/backend/src/main/kotlin/org/loculus/backend/config/ReleasedDataSpoolCleanupInterceptor.kt new file mode 100644 index 0000000000..ffa9a879c6 --- /dev/null +++ b/backend/src/main/kotlin/org/loculus/backend/config/ReleasedDataSpoolCleanupInterceptor.kt @@ -0,0 +1,47 @@ +package org.loculus.backend.config + +import jakarta.servlet.http.HttpServletRequest +import jakarta.servlet.http.HttpServletResponse +import org.loculus.backend.service.submission.SpooledStream +import org.springframework.web.context.request.NativeWebRequest +import org.springframework.web.context.request.RequestAttributes +import org.springframework.web.context.request.async.CallableProcessingInterceptor +import org.springframework.web.servlet.HandlerInterceptor +import java.util.concurrent.Callable + +internal const val RELEASED_DATA_SPOOL_ATTRIBUTE = "org.loculus.backend.releasedDataSpool" + +internal class ReleasedDataSpoolCleanupInterceptor : + HandlerInterceptor, + CallableProcessingInterceptor { + override fun afterCompletion( + request: HttpServletRequest, + response: HttpServletResponse, + handler: Any, + ex: Exception?, + ) { + closeReadySpool(request.getAttribute(RELEASED_DATA_SPOOL_ATTRIBUTE)) + } + + override fun postProcess(request: NativeWebRequest, task: Callable, concurrentResult: Any?) { + closeReadySpool(request.getAttribute(RELEASED_DATA_SPOOL_ATTRIBUTE, RequestAttributes.SCOPE_REQUEST)) + } + + override fun handleTimeout(request: NativeWebRequest, task: Callable): Any { + closeReadySpool(request.getAttribute(RELEASED_DATA_SPOOL_ATTRIBUTE, RequestAttributes.SCOPE_REQUEST)) + return CallableProcessingInterceptor.RESULT_NONE + } + + override fun handleError(request: NativeWebRequest, task: Callable, throwable: Throwable): Any { + closeReadySpool(request.getAttribute(RELEASED_DATA_SPOOL_ATTRIBUTE, RequestAttributes.SCOPE_REQUEST)) + return CallableProcessingInterceptor.RESULT_NONE + } + + override fun afterCompletion(request: NativeWebRequest, task: Callable) { + closeReadySpool(request.getAttribute(RELEASED_DATA_SPOOL_ATTRIBUTE, RequestAttributes.SCOPE_REQUEST)) + } + + private fun closeReadySpool(attribute: Any?) { + (attribute as? SpooledStream)?.closeIfTransferNotStarted() + } +} diff --git a/backend/src/main/kotlin/org/loculus/backend/config/SecurityConfig.kt b/backend/src/main/kotlin/org/loculus/backend/config/SecurityConfig.kt index f26d872343..2b4c2ced9d 100644 --- a/backend/src/main/kotlin/org/loculus/backend/config/SecurityConfig.kt +++ b/backend/src/main/kotlin/org/loculus/backend/config/SecurityConfig.kt @@ -13,6 +13,7 @@ import org.springframework.context.annotation.Configuration import org.springframework.core.convert.converter.Converter import org.springframework.http.HttpMethod import org.springframework.security.access.AccessDeniedException +import org.springframework.security.config.ObjectPostProcessor import org.springframework.security.config.annotation.web.builders.HttpSecurity import org.springframework.security.config.annotation.web.configuration.EnableWebSecurity import org.springframework.security.core.AuthenticationException @@ -28,6 +29,7 @@ import org.springframework.security.web.access.AccessDeniedHandler import org.springframework.security.web.access.AccessDeniedHandlerImpl import org.springframework.security.web.access.DelegatingAccessDeniedHandler import org.springframework.security.web.csrf.CsrfException +import org.springframework.security.web.header.HeaderWriterFilter import org.springframework.stereotype.Component private val log = KotlinLogging.logger { } @@ -80,6 +82,15 @@ class SecurityConfig { httpSecurity: HttpSecurity, keycloakAuthoritiesConverter: KeycloakAuthenticationConverter, ): SecurityFilterChain = httpSecurity + .headers { headers -> + headers.addObjectPostProcessor( + object : ObjectPostProcessor { + override fun postProcess(filter: O): O = filter.apply { + setShouldWriteHeadersEagerly(true) + } + }, + ) + } .authorizeHttpRequests { auth -> auth.requestMatchers( "/", diff --git a/backend/src/main/kotlin/org/loculus/backend/config/WebConfig.kt b/backend/src/main/kotlin/org/loculus/backend/config/WebConfig.kt index e30d5e9fa9..d3725d285e 100644 --- a/backend/src/main/kotlin/org/loculus/backend/config/WebConfig.kt +++ b/backend/src/main/kotlin/org/loculus/backend/config/WebConfig.kt @@ -4,12 +4,15 @@ import org.loculus.backend.auth.UserConverter import org.loculus.backend.log.OrganismMdcInterceptor import org.springframework.context.annotation.Configuration import org.springframework.web.method.support.HandlerMethodArgumentResolver +import org.springframework.web.servlet.config.annotation.AsyncSupportConfigurer import org.springframework.web.servlet.config.annotation.CorsRegistry import org.springframework.web.servlet.config.annotation.InterceptorRegistry import org.springframework.web.servlet.config.annotation.WebMvcConfigurer @Configuration class WebConfig(private val backendConfig: BackendConfig) : WebMvcConfigurer { + private val releasedDataSpoolCleanup = ReleasedDataSpoolCleanupInterceptor() + override fun addCorsMappings(registry: CorsRegistry) { registry.addMapping("/**") .allowedOrigins("*") // Allow requests from any origin @@ -21,6 +24,11 @@ class WebConfig(private val backendConfig: BackendConfig) : WebMvcConfigurer { override fun addInterceptors(registry: InterceptorRegistry) { registry.addInterceptor(ReadOnlyModeInterceptor(backendConfig)) registry.addInterceptor(OrganismMdcInterceptor()) + registry.addInterceptor(releasedDataSpoolCleanup) + } + + override fun configureAsyncSupport(configurer: AsyncSupportConfigurer) { + configurer.registerCallableInterceptors(releasedDataSpoolCleanup) } override fun addArgumentResolvers(resolvers: MutableList) { diff --git a/backend/src/main/kotlin/org/loculus/backend/controller/ExceptionHandler.kt b/backend/src/main/kotlin/org/loculus/backend/controller/ExceptionHandler.kt index 9da86b66ad..36a071f0f5 100644 --- a/backend/src/main/kotlin/org/loculus/backend/controller/ExceptionHandler.kt +++ b/backend/src/main/kotlin/org/loculus/backend/controller/ExceptionHandler.kt @@ -116,10 +116,16 @@ class ExceptionHandler : ResponseEntityExceptionHandler() { fun handleServiceUnavailableException(e: ServiceUnavailableException): ResponseEntity { log.info { "Caught service unavailable exception: ${e.message}" } - return responseEntity( - HttpStatus.SERVICE_UNAVAILABLE, - e.message, - ) + val response = ResponseEntity.status(HttpStatus.SERVICE_UNAVAILABLE) + e.retryAfterSeconds?.let { response.header(HttpHeaders.RETRY_AFTER, it.toString()) } + return response + .contentType(MediaType.APPLICATION_PROBLEM_JSON) + .body( + ProblemDetail.forStatusAndDetail( + HttpStatus.SERVICE_UNAVAILABLE, + e.message ?: HttpStatus.SERVICE_UNAVAILABLE.reasonPhrase, + ), + ) } private fun responseEntity(httpStatus: HttpStatus, detail: String?): ResponseEntity = @@ -171,4 +177,4 @@ class NotFoundException(message: String) : RuntimeException(message) class ProcessingValidationException(message: String) : RuntimeException(message) class DuplicateKeyException(message: String) : RuntimeException(message) class ConflictException(message: String) : RuntimeException(message) -class ServiceUnavailableException(message: String) : RuntimeException(message) +class ServiceUnavailableException(message: String, val retryAfterSeconds: Long? = null) : RuntimeException(message) diff --git a/backend/src/main/kotlin/org/loculus/backend/controller/SubmissionController.kt b/backend/src/main/kotlin/org/loculus/backend/controller/SubmissionController.kt index fc7a906fc7..a5433ce4a0 100644 --- a/backend/src/main/kotlin/org/loculus/backend/controller/SubmissionController.kt +++ b/backend/src/main/kotlin/org/loculus/backend/controller/SubmissionController.kt @@ -38,6 +38,7 @@ import org.loculus.backend.api.UnprocessedData import org.loculus.backend.auth.AuthenticatedUser import org.loculus.backend.auth.HiddenParam import org.loculus.backend.config.BackendConfig +import org.loculus.backend.config.RELEASED_DATA_SPOOL_ATTRIBUTE import org.loculus.backend.controller.LoculusCustomHeaders.X_TOTAL_RECORDS import org.loculus.backend.log.ORGANISM_MDC_KEY import org.loculus.backend.log.REQUEST_ID_MDC_KEY @@ -52,6 +53,8 @@ import org.loculus.backend.model.SubmissionParams import org.loculus.backend.model.SubmitModel import org.loculus.backend.service.datauseterms.DataUseTermsPreconditionValidator import org.loculus.backend.service.groupmanagement.GroupManagementPreconditionValidator +import org.loculus.backend.service.submission.SpooledStream +import org.loculus.backend.service.submission.StreamSpoolService import org.loculus.backend.service.submission.SubmissionDatabaseService import org.loculus.backend.utils.Accession import org.loculus.backend.utils.FastaEntry @@ -99,6 +102,7 @@ open class SubmissionController( private val releasedDataModel: ReleasedDataModel, private val submissionDatabaseService: SubmissionDatabaseService, private val iteratorStreamer: IteratorStreamer, + private val streamSpoolService: StreamSpoolService, private val requestIdContext: RequestIdContext, private val backendConfig: BackendConfig, private val objectMapper: ObjectMapper, @@ -321,6 +325,17 @@ open class SubmissionController( "No database changes since last request " + "(Etag in HttpHeaders.IF_NONE_MATCH matches lastDatabaseWriteETag)", ) + @ApiResponse( + responseCode = "503", + description = "The server cannot create another export yet", + headers = [ + Header( + name = "Retry-After", + description = "Seconds to wait before retrying", + schema = Schema(type = "integer"), + ), + ], + ) @GetMapping("/get-released-data", produces = [MediaType.APPLICATION_NDJSON_VALUE]) fun getReleasedData( @PathVariable @Valid organism: Organism, @@ -328,6 +343,7 @@ open class SubmissionController( @Parameter( description = "(Optional) Only retrieve all released data if Etag has changed.", ) @RequestHeader(value = HttpHeaders.IF_NONE_MATCH, required = false) ifNoneMatch: String?, + request: HttpServletRequest, ): ResponseEntity { val lastDatabaseWriteETag = releasedDataModel.getLastDatabaseWriteETag( tableNames = RELEASED_DATA_RELATED_TABLES, @@ -342,18 +358,21 @@ open class SubmissionController( headers.contentType = MediaType.APPLICATION_NDJSON compression?.let { headers.add(HttpHeaders.CONTENT_ENCODING, it.compressionName) } - val totalRecords = submissionDatabaseService.countReleasedSubmissions(organism) - 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, endpoint = "get-released-data", organism = organism) { + // Count while writing so the response header stays exact. + val spooled = streamSpoolService.spool(compression, endpoint = "get-released-data") { releasedDataModel.getReleasedData(organism) } - return ResponseEntity.ok().headers(headers).body(streamBody) + try { + request.setAttribute(RELEASED_DATA_SPOOL_ATTRIBUTE, spooled) + headers.add(X_TOTAL_RECORDS, spooled.recordCount.toString()) + headers.contentLength = spooled.file.length() + + return ResponseEntity.ok().headers(headers) + .body(spooledStreamBody(spooled, endpoint = "get-released-data", organism)) + } catch (error: Throwable) { + spooled.close() + throw error + } } @Operation(description = GET_DATA_TO_EDIT_SEQUENCE_VERSION_DESCRIPTION) @@ -464,11 +483,6 @@ open class SubmissionController( parsedAccessionVersions, ) 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, endpoint = "get-submitted-metadata", organism = organism) { submissionDatabaseService.streamSubmittedMetadata( @@ -673,6 +687,27 @@ open class SubmissionController( MDC.remove(ORGANISM_MDC_KEY) } + private fun spooledStreamBody(spooled: SpooledStream, endpoint: String, organism: Organism) = + StreamingResponseBody { responseBodyStream -> + val startTime = System.currentTimeMillis() + try { + check(spooled.beginTransfer()) { "The spooled response was closed before transfer started" } + MDC.put(REQUEST_ID_MDC_KEY, requestIdContext.requestId) + MDC.put(ORGANISM_MDC_KEY, organism.name) + spooled.file.inputStream().use { fileStream -> + fileStream.copyTo(responseBodyStream) + } + log.info { "[$endpoint] Response completed in ${System.currentTimeMillis() - startTime}ms" } + } catch (error: Exception) { + log.error(error) { "[$endpoint] Response failed after ${System.currentTimeMillis() - startTime}ms" } + throw error + } finally { + spooled.close() + MDC.remove(REQUEST_ID_MDC_KEY) + MDC.remove(ORGANISM_MDC_KEY) + } + } + fun parseFileMapping(fileMapping: String?, organism: Organism): SubmissionIdFilesMap? { val fileMappingParsed = fileMapping?.let { if (!backendConfig.getInstanceConfig(organism).schema.submissionDataTypes.files.enabled) { diff --git a/backend/src/main/kotlin/org/loculus/backend/service/submission/StreamSpoolService.kt b/backend/src/main/kotlin/org/loculus/backend/service/submission/StreamSpoolService.kt new file mode 100644 index 0000000000..8ffa206bf4 --- /dev/null +++ b/backend/src/main/kotlin/org/loculus/backend/service/submission/StreamSpoolService.kt @@ -0,0 +1,300 @@ +package org.loculus.backend.service.submission + +import jakarta.annotation.PreDestroy +import mu.KotlinLogging +import org.apache.commons.compress.compressors.zstandard.ZstdCompressorOutputStream +import org.jetbrains.exposed.sql.transactions.transaction +import org.loculus.backend.api.CompressionFormat +import org.loculus.backend.config.BackendSpringProperty +import org.loculus.backend.controller.ServiceUnavailableException +import org.loculus.backend.utils.IteratorStreamer +import org.springframework.beans.factory.annotation.Value +import org.springframework.scheduling.annotation.Scheduled +import org.springframework.stereotype.Service +import java.io.File +import java.io.FilterOutputStream +import java.io.OutputStream +import java.nio.channels.FileChannel +import java.nio.channels.FileLock +import java.nio.channels.OverlappingFileLockException +import java.nio.file.StandardOpenOption +import java.util.concurrent.ConcurrentHashMap +import java.util.concurrent.Semaphore +import java.util.concurrent.TimeUnit +import java.util.concurrent.atomic.AtomicReference + +private val log = KotlinLogging.logger { } + +const val SPOOL_FILE_PREFIX = "loculus-stream-" +const val SPOOL_RETRY_AFTER_SECONDS = 30L +private const val SPOOL_LOCK_FILE_NAME = ".loculus-spool.lock" +private const val SPOOL_PROBE_FILE_PREFIX = ".loculus-spool-probe-" +private enum class SpoolState { READY, TRANSFERRING, CLOSED } + +class SpooledStream internal constructor(val file: File, val recordCount: Long, private val release: () -> Unit) : + AutoCloseable { + private val state = AtomicReference(SpoolState.READY) + + internal fun beginTransfer(): Boolean = state.compareAndSet(SpoolState.READY, SpoolState.TRANSFERRING) + + internal fun closeIfTransferNotStarted() { + if (state.compareAndSet(SpoolState.READY, SpoolState.CLOSED)) { + release() + } + } + + override fun close() { + if (state.getAndSet(SpoolState.CLOSED) != SpoolState.CLOSED) { + release() + } + } +} + +/** Writes a database result to a temporary file and counts its records. */ +@Service +class StreamSpoolService( + private val iteratorStreamer: IteratorStreamer, + @Value("\${${BackendSpringProperty.STREAM_SPOOL_DIR}:}") spoolDir: String, + @Value("\${${BackendSpringProperty.STREAM_MAX_CONCURRENT_SPOOLS}:4}") maxConcurrentSpools: Int, + @Value("\${${BackendSpringProperty.STREAM_SPOOL_MAX_TOTAL_BYTES}:19327352832}") maxTotalBytes: Long, + @Value("\${${BackendSpringProperty.STREAM_SPOOL_MIN_FREE_BYTES}:1073741824}") minFreeBytes: Long, + @Value("\${${BackendSpringProperty.STREAM_SPOOL_FILE_TTL_MINUTES}:180}") private val spoolFileTtlMinutes: Long, +) { + private val usesDefaultTempDirectory = spoolDir.isBlank() + private val spoolDirectory = File(spoolDir.ifBlank { System.getProperty("java.io.tmpdir") }) + private val minFreeBytes = minFreeBytes.coerceAtLeast(0) + private val maxTotalBytes = maxTotalBytes.coerceAtLeast(1) + private val semaphore = Semaphore(maxConcurrentSpools.coerceAtLeast(1)) + private val activeFiles = ConcurrentHashMap.newKeySet() + private val diskWriteLock = Any() + private val trackedFileSizes = mutableMapOf() + private var trackedBytes = 0L + private var directoryLockChannel: FileChannel? = null + private var directoryLock: FileLock? = null + + init { + validateSpoolDirectory() + try { + if (usesDefaultTempDirectory) { + trackExistingFiles() + } else { + acquireDirectoryLock() + removeExistingFiles() + } + } catch (error: Throwable) { + releaseDirectoryLock() + throw error + } + } + + fun spool( + compressionFormat: CompressionFormat?, + endpoint: String, + sequenceProvider: () -> Sequence, + ): SpooledStream { + if (!semaphore.tryAcquire()) { + throw unavailable("The server is preparing the maximum number of exports. Please retry later.") + } + + val startTime = System.currentTimeMillis() + var tempFile: File? = null + try { + ensureFreeSpace() + val file = File.createTempFile("$SPOOL_FILE_PREFIX$endpoint-", ".ndjson", spoolDirectory) + tempFile = file + activeFiles.add(file) + synchronized(diskWriteLock) { + trackedFileSizes[file] = 0 + } + + val recordCount = transaction { + openOutputStream(file, compressionFormat).use { stream -> + iteratorStreamer.streamAsNdjson(sequenceProvider(), stream, flushPerRecord = false) + } + } + + log.info { + "[$endpoint] Spooled $recordCount records (${file.length()} bytes on disk) " + + "in ${System.currentTimeMillis() - startTime}ms" + } + return SpooledStream(file, recordCount) { + discard(file, endpoint) + } + } catch (error: Throwable) { + tempFile?.let { discard(it, endpoint) } + log.error(error) { + "[$endpoint] Failed to spool response after ${System.currentTimeMillis() - startTime}ms: $error" + } + throw error + } finally { + semaphore.release() + } + } + + private fun openOutputStream(file: File, compressionFormat: CompressionFormat?): OutputStream { + val fileStream = synchronized(diskWriteLock) { + val output = file.outputStream() + trackedBytes -= trackedFileSizes.put(file, 0) ?: 0 + output + } + val checkedFileStream = FreeSpaceCheckingOutputStream( + fileStream, + ) { bytes, write -> write(file, bytes, write) }.buffered() + return when (compressionFormat) { + CompressionFormat.ZSTD -> ZstdCompressorOutputStream(checkedFileStream) + null -> checkedFileStream + } + } + + private fun write(file: File, bytes: Int, write: () -> Unit) { + synchronized(diskWriteLock) { + ensureFreeSpace(bytes) + val requestedBytes = bytes.toLong() + if (requestedBytes > maxTotalBytes - trackedBytes) { + throw unavailable("The server does not have enough spool capacity. Please retry later.") + } + + trackedBytes += requestedBytes + trackedFileSizes[file] = trackedFileSizes.getValue(file) + requestedBytes + write() + } + } + + private fun ensureFreeSpace(bytesToWrite: Int = 0) { + val usable = spoolDirectory.usableSpace + val doesNotLeaveMinimum = usable <= minFreeBytes || bytesToWrite.toLong() > usable - minFreeBytes + if (usable == 0L || doesNotLeaveMinimum) { + throw unavailable("The server does not have enough free disk space. Please retry later.") + } + } + + private fun validateSpoolDirectory() { + check(spoolDirectory.isDirectory || spoolDirectory.mkdirs()) { + "Cannot create spool directory ${spoolDirectory.absolutePath}" + } + val probe = try { + File.createTempFile(SPOOL_PROBE_FILE_PREFIX, ".tmp", spoolDirectory) + } catch (error: Exception) { + throw IllegalStateException("Cannot write to spool directory ${spoolDirectory.absolutePath}", error) + } + check(probe.delete()) { + "Cannot delete files from spool directory ${spoolDirectory.absolutePath}" + } + } + + private fun acquireDirectoryLock() { + val channel = FileChannel.open( + File(spoolDirectory, SPOOL_LOCK_FILE_NAME).toPath(), + StandardOpenOption.CREATE, + StandardOpenOption.WRITE, + ) + try { + val lock = try { + channel.tryLock() + } catch (_: OverlappingFileLockException) { + null + } + check(lock != null) { + "Spool directory is already used by another process: ${spoolDirectory.absolutePath}" + } + directoryLockChannel = channel + directoryLock = lock + } catch (error: Throwable) { + channel.close() + throw error + } + } + + @PreDestroy + internal fun releaseDirectoryLock() { + try { + directoryLock?.release() + } catch (error: Exception) { + log.warn(error) { "Failed to release spool directory lock" } + } finally { + directoryLock = null + try { + directoryLockChannel?.close() + } catch (error: Exception) { + log.warn(error) { "Failed to close spool directory lock" } + } finally { + directoryLockChannel = null + } + } + } + + private fun trackExistingFiles() { + val existingFiles = spoolDirectory.listFiles { file -> + file.isFile && file.name.startsWith(SPOOL_FILE_PREFIX) + }.orEmpty() + synchronized(diskWriteLock) { + existingFiles.forEach { file -> trackedFileSizes[file] = file.length() } + trackedBytes = trackedFileSizes.values.sum() + } + } + + private fun removeExistingFiles() { + val existingFiles = spoolDirectory.listFiles { file -> + file.isFile && file.name.startsWith(SPOOL_FILE_PREFIX) + }.orEmpty() + existingFiles.forEach { file -> + check(file.delete() || !file.exists()) { + "Cannot remove spool file ${file.absolutePath}" + } + } + } + + private fun discard(file: File, endpoint: String) { + activeFiles.remove(file) + if (file.delete() || !file.exists()) { + synchronized(diskWriteLock) { + trackedBytes -= trackedFileSizes.remove(file) ?: 0 + } + } else { + log.warn { "[$endpoint] Failed to delete spool file ${file.absolutePath}" } + } + } + + private fun unavailable(message: String) = ServiceUnavailableException(message, SPOOL_RETRY_AFTER_SECONDS) + + @Scheduled( + initialDelay = 10, + fixedRateString = "\${${BackendSpringProperty.STREAM_SPOOL_SWEEP_EVERY_MINUTES}:30}", + timeUnit = TimeUnit.MINUTES, + ) + fun sweepOrphanedSpoolFiles() { + val cutoff = System.currentTimeMillis() - TimeUnit.MINUTES.toMillis(spoolFileTtlMinutes) + val orphans = spoolDirectory.listFiles { file -> + file.isFile && + file.name.startsWith(SPOOL_FILE_PREFIX) && + file !in activeFiles && + file.lastModified() < cutoff + } ?: return + val deleted = orphans.count { file -> + if (!file.delete()) { + false + } else { + synchronized(diskWriteLock) { + trackedBytes -= trackedFileSizes.remove(file) ?: 0 + } + true + } + } + if (deleted > 0) { + log.info { "Swept $deleted orphaned spool files from $spoolDirectory" } + } + } +} + +private class FreeSpaceCheckingOutputStream( + outputStream: OutputStream, + private val checkedWrite: (Int, () -> Unit) -> Unit, +) : FilterOutputStream(outputStream) { + override fun write(byte: Int) { + checkedWrite(1) { out.write(byte) } + } + + override fun write(bytes: ByteArray, offset: Int, length: Int) { + checkedWrite(length) { out.write(bytes, offset, length) } + } +} 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 index 057ca2cf3b..f096f97c8b 100644 --- a/backend/src/main/kotlin/org/loculus/backend/service/submission/SubmissionDatabaseService.kt +++ b/backend/src/main/kotlin/org/loculus/backend/service/submission/SubmissionDatabaseService.kt @@ -757,21 +757,6 @@ class SubmissionDatabaseService( return result } - // Make sure to keep in sync with streamReleasedSubmissions query - fun countReleasedSubmissions(organism: Organism): Long { - val startTime = dateProvider.getCurrentInstant() - val result = SequenceEntriesView.select( - SequenceEntriesView.accessionColumn, - ).where { - SequenceEntriesView.statusIs(Status.APPROVED_FOR_RELEASE) and SequenceEntriesView.organismIs( - organism, - ) - }.count() - log.info { "Counting released submissions for $organism took ${durationTillNowInMs(startTime)} ms" } - return result - } - - // Make sure to keep in sync with countReleasedSubmissions query fun streamReleasedSubmissions(organism: Organism): Sequence = SequenceEntriesView.join( DataUseTermsTable, JoinType.LEFT, diff --git a/backend/src/main/kotlin/org/loculus/backend/utils/IteratorStreamer.kt b/backend/src/main/kotlin/org/loculus/backend/utils/IteratorStreamer.kt index e8cf7f1b74..8a881464e2 100644 --- a/backend/src/main/kotlin/org/loculus/backend/utils/IteratorStreamer.kt +++ b/backend/src/main/kotlin/org/loculus/backend/utils/IteratorStreamer.kt @@ -6,15 +6,21 @@ import java.io.OutputStream @Service class IteratorStreamer(private val objectMapper: ObjectMapper) { - fun streamAsNdjson(sequence: Sequence, outputStream: OutputStream) = - streamAsNdjson(sequence.iterator(), outputStream) + /** Writes NDJSON, returns the record count, and can skip flushing each record. */ + fun streamAsNdjson(sequence: Sequence, outputStream: OutputStream, flushPerRecord: Boolean = true): Long = + streamAsNdjson(sequence.iterator(), outputStream, flushPerRecord) - fun streamAsNdjson(iterator: Iterator, outputStream: OutputStream) { + fun streamAsNdjson(iterator: Iterator, outputStream: OutputStream, flushPerRecord: Boolean = true): Long { + var count = 0L iterator.forEach { val json = objectMapper.writeValueAsString(it) outputStream.write(json.toByteArray()) outputStream.write('\n'.code) - outputStream.flush() + if (flushPerRecord) { + outputStream.flush() + } + count++ } + return count } } diff --git a/backend/src/main/resources/application.properties b/backend/src/main/resources/application.properties index 8cb9cba040..697242c6ec 100644 --- a/backend/src/main/resources/application.properties +++ b/backend/src/main/resources/application.properties @@ -32,6 +32,17 @@ loculus.pipeline-version-upgrade-check.interval-seconds=10 loculus.seqset-citations.run-every-minutes=360 loculus.maintenance.clean-up-aux-table.run-every-hours=1 loculus.stream.batch-size=1000 +# Dedicated directory for released data exports. Empty uses the JVM temp directory. +loculus.stream.spool-dir= +# Maximum number of released data exports prepared at once. +loculus.stream.max-concurrent-spools=4 +# Total space available to spool files. +loculus.stream.spool-max-total-bytes=19327352832 +# Free space that must remain in the spool directory. +loculus.stream.spool-min-free-bytes=1073741824 +# Age when an abandoned spool file can be removed. +loculus.stream.spool-file-ttl-minutes=180 +loculus.stream.spool-sweep-every-minutes=30 loculus.debug-mode=false loculus.s3.enabled=false diff --git a/backend/src/test/kotlin/org/loculus/backend/config/ReleasedDataSpoolCleanupInterceptorTest.kt b/backend/src/test/kotlin/org/loculus/backend/config/ReleasedDataSpoolCleanupInterceptorTest.kt new file mode 100644 index 0000000000..2516615552 --- /dev/null +++ b/backend/src/test/kotlin/org/loculus/backend/config/ReleasedDataSpoolCleanupInterceptorTest.kt @@ -0,0 +1,91 @@ +package org.loculus.backend.config + +import org.hamcrest.MatcherAssert.assertThat +import org.hamcrest.Matchers.`is` +import org.junit.jupiter.api.Test +import org.loculus.backend.service.submission.SpooledStream +import org.springframework.mock.web.MockHttpServletRequest +import org.springframework.mock.web.MockHttpServletResponse +import org.springframework.web.context.request.ServletWebRequest +import java.io.File +import java.util.concurrent.Callable + +class ReleasedDataSpoolCleanupInterceptorTest { + @Test + fun `after completion closes a spool before transfer starts`() { + var closeCount = 0 + val spooled = SpooledStream(File("unused"), 0) { closeCount++ } + val request = MockHttpServletRequest().apply { + setAttribute(RELEASED_DATA_SPOOL_ATTRIBUTE, spooled) + } + val interceptor = ReleasedDataSpoolCleanupInterceptor() + + interceptor.afterCompletion(request, MockHttpServletResponse(), Any(), null) + interceptor.afterCompletion(request, MockHttpServletResponse(), Any(), null) + spooled.close() + + assertThat(closeCount, `is`(1)) + assertThat(spooled.beginTransfer(), `is`(false)) + } + + @Test + fun `after completion leaves a transferring spool open`() { + var closeCount = 0 + val spooled = SpooledStream(File("unused"), 0) { closeCount++ } + val request = MockHttpServletRequest().apply { + setAttribute(RELEASED_DATA_SPOOL_ATTRIBUTE, spooled) + } + + assertThat(spooled.beginTransfer(), `is`(true)) + val interceptor = ReleasedDataSpoolCleanupInterceptor() + val webRequest = ServletWebRequest(request) + val task = Callable { Unit } + + interceptor.handleTimeout(webRequest, task) + interceptor.handleError(webRequest, task, RuntimeException("disconnected")) + interceptor.postProcess(webRequest, task, Unit) + interceptor.afterCompletion(webRequest, task) + interceptor.afterCompletion(request, MockHttpServletResponse(), Any(), null) + assertThat(closeCount, `is`(0)) + + spooled.close() + assertThat(closeCount, `is`(1)) + } + + @Test + fun `async terminal callbacks close a spool before transfer starts`() { + val interceptor = ReleasedDataSpoolCleanupInterceptor() + val task = Callable { Unit } + + var timeoutCloseCount = 0 + val timeoutSpool = SpooledStream(File("unused"), 0) { timeoutCloseCount++ } + interceptor.handleTimeout(webRequestFor(timeoutSpool), task) + + var errorCloseCount = 0 + val errorSpool = SpooledStream(File("unused"), 0) { errorCloseCount++ } + interceptor.handleError(webRequestFor(errorSpool), task, RuntimeException("disconnected")) + + var postProcessCloseCount = 0 + val postProcessSpool = SpooledStream(File("unused"), 0) { postProcessCloseCount++ } + interceptor.postProcess(webRequestFor(postProcessSpool), task, RuntimeException("rejected")) + + var completionCloseCount = 0 + val completionSpool = SpooledStream(File("unused"), 0) { completionCloseCount++ } + interceptor.afterCompletion(webRequestFor(completionSpool), task) + + assertThat(timeoutCloseCount, `is`(1)) + assertThat(errorCloseCount, `is`(1)) + assertThat(postProcessCloseCount, `is`(1)) + assertThat(completionCloseCount, `is`(1)) + assertThat(timeoutSpool.beginTransfer(), `is`(false)) + assertThat(errorSpool.beginTransfer(), `is`(false)) + assertThat(postProcessSpool.beginTransfer(), `is`(false)) + assertThat(completionSpool.beginTransfer(), `is`(false)) + } + + private fun webRequestFor(spooled: SpooledStream) = ServletWebRequest( + MockHttpServletRequest().apply { + setAttribute(RELEASED_DATA_SPOOL_ATTRIBUTE, spooled) + }, + ) +} diff --git a/backend/src/test/kotlin/org/loculus/backend/controller/TestHelpers.kt b/backend/src/test/kotlin/org/loculus/backend/controller/TestHelpers.kt index d69ccc300f..d4f96622b1 100644 --- a/backend/src/test/kotlin/org/loculus/backend/controller/TestHelpers.kt +++ b/backend/src/test/kotlin/org/loculus/backend/controller/TestHelpers.kt @@ -84,7 +84,10 @@ inline fun ResultActions.expectNdjsonAndGetContent(): List { } fun awaitResponse(result: MvcResult): String { - result.getAsyncResult() + val asyncResult = result.getAsyncResult() + if (asyncResult is Throwable) { + throw asyncResult + } return result.response.contentAsString } diff --git a/backend/src/test/kotlin/org/loculus/backend/controller/submission/GetReleasedDataEndpointTest.kt b/backend/src/test/kotlin/org/loculus/backend/controller/submission/GetReleasedDataEndpointTest.kt index d4a680655f..bac8a9f2ab 100644 --- a/backend/src/test/kotlin/org/loculus/backend/controller/submission/GetReleasedDataEndpointTest.kt +++ b/backend/src/test/kotlin/org/loculus/backend/controller/submission/GetReleasedDataEndpointTest.kt @@ -116,6 +116,37 @@ class GetReleasedDataEndpointTest( .andExpect(header().string("x-total-records", `is`("0"))) } + @Test + fun `GIVEN no sequence entries WHEN requesting zstd THEN returns a valid empty compressed response`() { + val response = submissionControllerClient.getReleasedData(compression = "zstd") + .andExpect(status().isOk) + .andExpect(header().string(HttpHeaders.CONTENT_ENCODING, "zstd")) + .andExpect(header().string("x-total-records", `is`("0"))) + .andReturn() + response.getAsyncResult() + + val decompressedContent = ZstdInputStream(response.response.contentAsByteArray.inputStream()) + .apply { continuous = true } + .readAllBytes() + .decodeToString() + + assertThat(decompressedContent, `is`("")) + } + + @Test + fun `GIVEN released data THEN sets a Content-Length header matching the served body`() { + convenienceClient.prepareDefaultSequenceEntriesToApprovedForRelease() + + val response = submissionControllerClient.getReleasedData() + .andExpect(status().isOk) + .andReturn() + response.getAsyncResult() + + val bodyLength = response.response.contentAsByteArray.size + assertThat(bodyLength, greaterThan(0)) + assertThat(response.response.getHeader(HttpHeaders.CONTENT_LENGTH), `is`(bodyLength.toString())) + } + @Test fun `Given released data THEN does not have unknown top-level fields`() { val allowedKeys = setOf( diff --git a/backend/src/test/kotlin/org/loculus/backend/controller/submission/GetReleasedDataSpoolEndpointTest.kt b/backend/src/test/kotlin/org/loculus/backend/controller/submission/GetReleasedDataSpoolEndpointTest.kt new file mode 100644 index 0000000000..459797acc6 --- /dev/null +++ b/backend/src/test/kotlin/org/loculus/backend/controller/submission/GetReleasedDataSpoolEndpointTest.kt @@ -0,0 +1,160 @@ +package org.loculus.backend.controller.submission + +import com.ninjasquad.springmockk.MockkBean +import io.mockk.every +import org.hamcrest.MatcherAssert.assertThat +import org.hamcrest.Matchers.`is` +import org.junit.jupiter.api.BeforeEach +import org.junit.jupiter.api.Test +import org.junit.jupiter.api.assertThrows +import org.keycloak.representations.idm.UserRepresentation +import org.loculus.backend.api.ReleasedData +import org.loculus.backend.controller.EndpointTest +import org.loculus.backend.controller.ServiceUnavailableException +import org.loculus.backend.controller.expectNdjsonAndGetContent +import org.loculus.backend.service.KeycloakAdapter +import org.loculus.backend.service.submission.SPOOL_FILE_PREFIX +import org.loculus.backend.service.submission.SPOOL_RETRY_AFTER_SECONDS +import org.loculus.backend.service.submission.SpooledStream +import org.loculus.backend.service.submission.StreamSpoolService +import org.loculus.backend.utils.IteratorStreamer +import org.springframework.beans.factory.annotation.Autowired +import org.springframework.http.HttpHeaders +import org.springframework.test.web.servlet.result.MockMvcResultMatchers.status +import java.io.File +import java.sql.SQLException +import java.util.concurrent.CountDownLatch +import java.util.concurrent.Executors +import java.util.concurrent.TimeUnit + +private const val TEST_SPOOL_DIR = "build/test-released-data-spool" + +@EndpointTest( + properties = [ + "loculus.stream.spool-dir=$TEST_SPOOL_DIR", + "loculus.stream.max-concurrent-spools=1", + ], +) +class GetReleasedDataSpoolEndpointTest( + @Autowired private val convenienceClient: SubmissionConvenienceClient, + @Autowired private val submissionControllerClient: SubmissionControllerClient, + @Autowired private val streamSpoolService: StreamSpoolService, + @Autowired private val iteratorStreamer: IteratorStreamer, +) { + private val spoolDir = File(TEST_SPOOL_DIR) + + @MockkBean + lateinit var keycloakAdapter: KeycloakAdapter + + @BeforeEach + fun setup() { + every { keycloakAdapter.getUsersWithName(any()) } returns listOf(UserRepresentation()) + spoolFiles().forEach { it.delete() } + } + + private fun spoolFiles(): List = + spoolDir.listFiles { file -> file.name.startsWith(SPOOL_FILE_PREFIX) }?.toList().orEmpty() + + @Test + fun `GIVEN released data THEN the spool temp file is deleted after the response is served`() { + convenienceClient.prepareDefaultSequenceEntriesToApprovedForRelease() + + val responseBody = submissionControllerClient.getReleasedData() + .andExpect(status().isOk) + .expectNdjsonAndGetContent() + assertThat(responseBody.isNotEmpty(), `is`(true)) + + assertThat(spoolFiles(), `is`(emptyList())) + } + + @Test + fun `GIVEN a database retry THEN the spool contains only the successful attempt`() { + var attempts = 0 + + val spooled = streamSpoolService.spool(null, endpoint = "retry-test") { + val attempt = ++attempts + sequence { + yield("attempt $attempt") + if (attempt == 1) { + throw SQLException("retry this transaction") + } + } + } + + try { + assertThat(attempts, `is`(2)) + assertThat(spooled.recordCount, `is`(1L)) + assertThat(spooled.file.readLines(), `is`(listOf("\"attempt 2\""))) + } finally { + spooled.close() + } + } + + @Test + fun `GIVEN generation in progress THEN another request gets retry guidance`() { + val generationStarted = CountDownLatch(1) + val finishGeneration = CountDownLatch(1) + val executor = Executors.newSingleThreadExecutor() + val generation = executor.submit { + streamSpoolService.spool(null, endpoint = "active-test") { + sequence { + generationStarted.countDown() + check(finishGeneration.await(10, TimeUnit.SECONDS)) + yield("complete") + } + } + } + try { + assertThat(generationStarted.await(10, TimeUnit.SECONDS), `is`(true)) + + submissionControllerClient.getReleasedData() + .andExpect(status().isServiceUnavailable) + .andExpect { + assertThat( + it.response.getHeader(HttpHeaders.RETRY_AFTER), + `is`(SPOOL_RETRY_AFTER_SECONDS.toString()), + ) + } + + finishGeneration.countDown() + val completed = generation.get(10, TimeUnit.SECONDS) + + completed.file.setLastModified( + System.currentTimeMillis() - TimeUnit.DAYS.toMillis(1), + ) + streamSpoolService.sweepOrphanedSpoolFiles() + assertThat(completed.file.exists(), `is`(true)) + + streamSpoolService.spool(null, endpoint = "next-test") { emptySequence() }.close() + } finally { + finishGeneration.countDown() + runCatching { generation.get(10, TimeUnit.SECONDS).close() } + executor.shutdownNow() + } + } + + @Test + fun `GIVEN a full spool quota THEN writing stops with a controlled error`() { + val limitedDir = File(spoolDir, "limited") + val limitedService = StreamSpoolService( + iteratorStreamer = iteratorStreamer, + spoolDir = limitedDir.absolutePath, + maxConcurrentSpools = 1, + maxTotalBytes = 1, + minFreeBytes = 0, + spoolFileTtlMinutes = 60, + ) + + try { + assertThrows { + limitedService.spool(null, endpoint = "quota-test") { sequenceOf("too large") } + } + assertThat( + limitedDir.listFiles { file -> file.name.startsWith(SPOOL_FILE_PREFIX) }?.toList().orEmpty(), + `is`(emptyList()), + ) + } finally { + limitedService.releaseDirectoryLock() + } + } +} diff --git a/backend/src/test/kotlin/org/loculus/backend/service/submission/StreamSpoolServiceTest.kt b/backend/src/test/kotlin/org/loculus/backend/service/submission/StreamSpoolServiceTest.kt new file mode 100644 index 0000000000..ccf4666505 --- /dev/null +++ b/backend/src/test/kotlin/org/loculus/backend/service/submission/StreamSpoolServiceTest.kt @@ -0,0 +1,98 @@ +package org.loculus.backend.service.submission + +import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper +import org.hamcrest.MatcherAssert.assertThat +import org.hamcrest.Matchers.`is` +import org.junit.jupiter.api.AfterEach +import org.junit.jupiter.api.Test +import org.junit.jupiter.api.assertThrows +import org.junit.jupiter.api.io.TempDir +import org.loculus.backend.api.CompressionFormat +import org.loculus.backend.controller.ServiceUnavailableException +import org.loculus.backend.utils.IteratorStreamer +import java.io.File +import java.util.concurrent.TimeUnit + +class StreamSpoolServiceTest { + private val iteratorStreamer = IteratorStreamer(jacksonObjectMapper()) + private val services = mutableListOf() + + private fun serviceFor(dir: File, minFreeBytes: Long = 0, ttlMinutes: Long = 60) = StreamSpoolService( + iteratorStreamer = iteratorStreamer, + spoolDir = dir.absolutePath, + maxConcurrentSpools = 4, + maxTotalBytes = Long.MAX_VALUE, + minFreeBytes = minFreeBytes, + spoolFileTtlMinutes = ttlMinutes, + ).also { services.add(it) } + + @AfterEach + fun releaseDirectoryLocks() { + services.forEach { it.releaseDirectoryLock() } + } + + private fun spoolFilesIn(dir: File): List = + dir.listFiles { file -> file.name.startsWith(SPOOL_FILE_PREFIX) }?.toList().orEmpty() + + @Test + fun `spool rejects with a 503 exception when free disk space is below the threshold`(@TempDir dir: File) { + val service = serviceFor(dir, minFreeBytes = Long.MAX_VALUE) + + assertThrows { + service.spool(CompressionFormat.ZSTD, endpoint = "get-released-data") { sequenceOf("a", "b") } + } + + assertThat(spoolFilesIn(dir), `is`(emptyList())) + } + + @Test + fun `service rejects a spool path that is a file`(@TempDir dir: File) { + val file = File(dir, "not-a-directory").apply { writeText("content") } + + assertThrows { serviceFor(file) } + } + + @Test + fun `service removes old spool files from an explicit directory at startup`(@TempDir dir: File) { + val oldSpool = File(dir, "${SPOOL_FILE_PREFIX}old.ndjson").apply { writeText("old") } + val unrelated = File(dir, "unrelated.ndjson").apply { writeText("keep") } + + serviceFor(dir) + + assertThat(oldSpool.exists(), `is`(false)) + assertThat(unrelated.exists(), `is`(true)) + } + + @Test + fun `explicit spool directory has one owner`(@TempDir dir: File) { + val first = serviceFor(dir) + + assertThrows { serviceFor(dir) } + + first.releaseDirectoryLock() + serviceFor(dir) + } + + @Test + fun `sweepOrphanedSpoolFiles deletes stale spool files but keeps fresh and unrelated ones`(@TempDir dir: File) { + val ttlMinutes = 60L + val service = serviceFor(dir, ttlMinutes = ttlMinutes) + val staleMillis = System.currentTimeMillis() - TimeUnit.MINUTES.toMillis(ttlMinutes * 2) + + val stale = File(dir, "${SPOOL_FILE_PREFIX}get-released-data-stale.ndjson").apply { + writeText("stale") + setLastModified(staleMillis) + } + val fresh = File(dir, "${SPOOL_FILE_PREFIX}get-released-data-fresh.ndjson").apply { writeText("fresh") } + val unrelated = File(dir, "some-other-file.ndjson").apply { + writeText("unrelated") + setLastModified(staleMillis) + } + + service.sweepOrphanedSpoolFiles() + + assertThat(stale.exists(), `is`(false)) + assertThat(fresh.exists(), `is`(true)) + assertThat(unrelated.exists(), `is`(true)) + } +} diff --git a/backend/src/test/kotlin/org/loculus/backend/utils/IteratorStreamerTest.kt b/backend/src/test/kotlin/org/loculus/backend/utils/IteratorStreamerTest.kt new file mode 100644 index 0000000000..b170cc21d5 --- /dev/null +++ b/backend/src/test/kotlin/org/loculus/backend/utils/IteratorStreamerTest.kt @@ -0,0 +1,45 @@ +package org.loculus.backend.utils + +import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper +import org.hamcrest.MatcherAssert.assertThat +import org.hamcrest.Matchers.`is` +import org.junit.jupiter.api.Test +import java.io.ByteArrayOutputStream + +class IteratorStreamerTest { + private val streamer = IteratorStreamer(jacksonObjectMapper()) + + @Test + fun `returns the number of records written and one ndjson line per record`() { + val output = ByteArrayOutputStream() + + val count = streamer.streamAsNdjson(sequenceOf("a", "b", "c"), output) + + assertThat(count, `is`(3L)) + val lines = output.toString(Charsets.UTF_8).lines().filter { it.isNotEmpty() } + assertThat(lines, `is`(listOf("\"a\"", "\"b\"", "\"c\""))) + } + + @Test + fun `returns zero and writes nothing for an empty sequence`() { + val output = ByteArrayOutputStream() + + val count = streamer.streamAsNdjson(emptySequence(), output) + + assertThat(count, `is`(0L)) + assertThat(output.size(), `is`(0)) + } + + @Test + fun `counts records identically whether or not it flushes per record`() { + val flushed = ByteArrayOutputStream() + val buffered = ByteArrayOutputStream() + + val flushedCount = streamer.streamAsNdjson(sequenceOf(1, 2, 3, 4), flushed, flushPerRecord = true) + val bufferedCount = streamer.streamAsNdjson(sequenceOf(1, 2, 3, 4), buffered, flushPerRecord = false) + + assertThat(flushedCount, `is`(4L)) + assertThat(bufferedCount, `is`(4L)) + assertThat(buffered.toByteArray(), `is`(flushed.toByteArray())) + } +} diff --git a/cli/src/loculus_cli/api/backend.py b/cli/src/loculus_cli/api/backend.py index d5f9d79405..eabd56e0e2 100644 --- a/cli/src/loculus_cli/api/backend.py +++ b/cli/src/loculus_cli/api/backend.py @@ -15,6 +15,13 @@ UnprocessedData, ) +DEFAULT_HTTP_TIMEOUT_SECONDS = 30.0 +RELEASED_DATA_READ_TIMEOUT_SECONDS = 3600.0 +RELEASED_DATA_TIMEOUT = httpx.Timeout( + DEFAULT_HTTP_TIMEOUT_SECONDS, + read=RELEASED_DATA_READ_TIMEOUT_SECONDS, +) + class BackendClient: """Client for Loculus backend API.""" @@ -24,7 +31,7 @@ def __init__(self, instance_config: InstanceConfig, auth_client: AuthClient): self.auth_client = auth_client self.client = httpx.Client( base_url=instance_config.backend_url, - timeout=30.0, + timeout=DEFAULT_HTTP_TIMEOUT_SECONDS, follow_redirects=True, ) @@ -198,7 +205,11 @@ def get_released_data( } try: - response = self.client.get(f"/{organism}/get-released-data", params=params) + response = self.client.get( + f"/{organism}/get-released-data", + params=params, + timeout=RELEASED_DATA_TIMEOUT, + ) response.raise_for_status() return response.content except httpx.HTTPStatusError as e: diff --git a/cli/tests/test_backend.py b/cli/tests/test_backend.py new file mode 100644 index 0000000000..9954767a50 --- /dev/null +++ b/cli/tests/test_backend.py @@ -0,0 +1,27 @@ +from __future__ import annotations + +from unittest.mock import Mock + +from loculus_cli.api.backend import ( + DEFAULT_HTTP_TIMEOUT_SECONDS, + RELEASED_DATA_READ_TIMEOUT_SECONDS, + BackendClient, +) + + +def test_get_released_data_uses_long_read_timeout() -> None: + client = BackendClient.__new__(BackendClient) + http_client = Mock() + client.client = http_client + response = Mock() + response.content = b"released data" + http_client.get.return_value = response + + result = client.get_released_data("mpox") + + assert result == b"released data" + timeout = http_client.get.call_args.kwargs["timeout"] + assert timeout.connect == DEFAULT_HTTP_TIMEOUT_SECONDS + assert timeout.read == RELEASED_DATA_READ_TIMEOUT_SECONDS + assert timeout.write == DEFAULT_HTTP_TIMEOUT_SECONDS + assert timeout.pool == DEFAULT_HTTP_TIMEOUT_SECONDS diff --git a/docs/src/content/docs/for-administrators/setup-with-k3d-and-nginx.mdx b/docs/src/content/docs/for-administrators/setup-with-k3d-and-nginx.mdx index 3403a81032..4a1001e856 100644 --- a/docs/src/content/docs/for-administrators/setup-with-k3d-and-nginx.mdx +++ b/docs/src/content/docs/for-administrators/setup-with-k3d-and-nginx.mdx @@ -280,6 +280,7 @@ server { proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; proxy_set_header X-Forwarded-Proto $scheme; proxy_set_header X-Forwarded-Prefix /backend/; + proxy_read_timeout 3600s; } location / { diff --git a/kubernetes/loculus/templates/ena-submission-deployment.yaml b/kubernetes/loculus/templates/ena-submission-deployment.yaml index 92fee97f8c..7adc378ae9 100644 --- a/kubernetes/loculus/templates/ena-submission-deployment.yaml +++ b/kubernetes/loculus/templates/ena-submission-deployment.yaml @@ -127,7 +127,7 @@ kind: CronJob metadata: name: loculus-get-ena-submission-list-cronjob spec: - # run twice a day to ensure at least once no overlap with argo cd refresh + # Run twice daily so one run can avoid an Argo CD refresh. schedule: "0 1,13 * * *" startingDeadlineSeconds: 60 concurrencyPolicy: Forbid diff --git a/kubernetes/loculus/templates/loculus-backend.yaml b/kubernetes/loculus/templates/loculus-backend.yaml index 1252425162..5a536b0307 100644 --- a/kubernetes/loculus/templates/loculus-backend.yaml +++ b/kubernetes/loculus/templates/loculus-backend.yaml @@ -70,6 +70,12 @@ spec: - "--spring.security.oauth2.resourceserver.jwt.jwk-set-uri=http://loculus-keycloak-service:8083/realms/loculus/protocol/openid-connect/certs" - "--loculus.cleanup.task.reset-stale-in-processing-after-seconds={{- .Values.preprocessingTimeout | default 120 }}" - "--loculus.pipeline-version-upgrade-check.interval-seconds={{- .Values.pipelineVersionUpgradeCheckIntervalSeconds | default 10 }}" + - "--loculus.stream.spool-dir={{ $.Values.backendStreamSpool.directory }}" + - "--loculus.stream.spool-max-total-bytes={{ int64 $.Values.backendStreamSpool.maxTotalBytes }}" + - "--loculus.stream.max-concurrent-spools={{ $.Values.backendStreamSpool.maxConcurrentSpools }}" + - "--loculus.stream.spool-min-free-bytes={{ int64 $.Values.backendStreamSpool.minFreeBytes }}" + - "--loculus.stream.spool-file-ttl-minutes={{ $.Values.backendStreamSpool.fileTtlMinutes }}" + - "--loculus.stream.spool-sweep-every-minutes={{ $.Values.backendStreamSpool.sweepEveryMinutes }}" - "--loculus.s3.enabled=$(S3_ENABLED)" {{- if $.Values.s3.enabled }} - "--loculus.s3.gc-enabled=$(S3_GC_ENABLED)" @@ -166,6 +172,11 @@ spec: volumeMounts: - name: loculus-backend-config-processed mountPath: /config + - name: loculus-backend-spool + mountPath: {{ $.Values.backendStreamSpool.directory | quote }} volumes: {{ include "loculus.configVolume" (dict "name" "loculus-backend-config") | nindent 8 }} + - name: loculus-backend-spool + emptyDir: + sizeLimit: {{ $.Values.backendStreamSpool.sizeLimit | quote }} {{- end }} diff --git a/kubernetes/loculus/templates/silo-deployment.yaml b/kubernetes/loculus/templates/silo-deployment.yaml index b3fa005d91..d00be82dbd 100644 --- a/kubernetes/loculus/templates/silo-deployment.yaml +++ b/kubernetes/loculus/templates/silo-deployment.yaml @@ -12,10 +12,10 @@ kind: Deployment metadata: name: loculus-silo-{{ $key }} annotations: - # Force replace when run a) with dev db and b) without persistence to ensure silo prepro fails loudly + # Recreate SILO for disposable databases so import failures stay visible. argocd.argoproj.io/sync-options: Replace=true{{ if (and (not $.Values.developmentDatabasePersistence) $.Values.runDevelopmentMainDatabase) }},Force=true{{ end }} spec: - progressDeadlineSeconds: 1200 # 20 minute timeout to allow for long silo imports e.g. mpox + progressDeadlineSeconds: {{ $.Values.siloImport.progressDeadlineSeconds }} replicas: 1 selector: matchLabels: @@ -106,6 +106,8 @@ spec: value: {{ $.Values.siloImport.hardRefreshIntervalSeconds | quote }} - name: SILO_IMPORT_POLL_INTERVAL_SECONDS value: {{ $.Values.siloImport.pollIntervalSeconds | quote }} + - name: SILO_IMPORT_DOWNLOAD_TIMEOUT_SECONDS + value: {{ $.Values.siloImport.downloadTimeoutSeconds | quote }} - name: PATH_TO_SILO_BINARY value: "/usr/local/bin/silo" - name: PREPROCESSING_CONFIG diff --git a/kubernetes/loculus/values.schema.json b/kubernetes/loculus/values.schema.json index 134655edc6..1dbebff96a 100644 --- a/kubernetes/loculus/values.schema.json +++ b/kubernetes/loculus/values.schema.json @@ -1819,7 +1819,9 @@ }, "getSubmissionListLimitSeconds": { "type": "integer", - "default": 600 + "description": "Runtime limit for the ENA submission list job. It covers download and processing.", + "default": 10800, + "minimum": 3600 }, "replicas": { "type": "object", @@ -1948,13 +1950,94 @@ "description": "Interval in seconds between checking for new data", "default": 30, "minimum": 1 + }, + "downloadTimeoutSeconds": { + "type": "integer", + "description": "Read timeout for downloading released data in seconds", + "default": 3600, + "minimum": 1 + }, + "progressDeadlineSeconds": { + "type": "integer", + "description": "SILO rollout deadline in seconds", + "default": 10800, + "minimum": 1 } }, "additionalProperties": false, "default": { "siloTimeoutSeconds": 3600, "hardRefreshIntervalSeconds": 3600, - "pollIntervalSeconds": 30 + "pollIntervalSeconds": 30, + "downloadTimeoutSeconds": 3600, + "progressDeadlineSeconds": 10800 + } + }, + "backendStreamSpool": { + "type": "object", + "description": "Storage and limits for backend export files", + "properties": { + "directory": { + "type": "string", + "description": "Path mounted for backend export files", + "default": "/var/lib/loculus/spool", + "pattern": "^/.+" + }, + "sizeLimit": { + "type": "string", + "description": "Maximum size of the export volume", + "default": "20Gi", + "minLength": 1 + }, + "maxTotalBytes": { + "type": "integer", + "description": "Maximum bytes used by backend export files. Keep this below the volume size limit.", + "default": 19327352832, + "minimum": 1 + }, + "maxConcurrentSpools": { + "type": "integer", + "description": "Maximum number of exports prepared at once", + "default": 4, + "minimum": 1 + }, + "minFreeBytes": { + "type": "integer", + "description": "Free bytes kept available on the export volume", + "default": 1073741824, + "minimum": 0 + }, + "fileTtlMinutes": { + "type": "integer", + "description": "Age when abandoned export files are removed", + "default": 180, + "minimum": 1 + }, + "sweepEveryMinutes": { + "type": "integer", + "description": "Interval for removing abandoned export files", + "default": 30, + "minimum": 1 + } + }, + "additionalProperties": false, + "required": [ + "directory", + "sizeLimit", + "maxTotalBytes", + "maxConcurrentSpools", + "minFreeBytes", + "fileTtlMinutes", + "sweepEveryMinutes" + ], + "default": { + "directory": "/var/lib/loculus/spool", + "sizeLimit": "20Gi", + "maxTotalBytes": 19327352832, + "maxConcurrentSpools": 4, + "minFreeBytes": 1073741824, + "fileTtlMinutes": 180, + "sweepEveryMinutes": 30 } }, "gitHubEditLink": { diff --git a/kubernetes/loculus/values.yaml b/kubernetes/loculus/values.yaml index 62ef79196f..0616bf72ec 100644 --- a/kubernetes/loculus/values.yaml +++ b/kubernetes/loculus/values.yaml @@ -47,8 +47,21 @@ siloImport: siloTimeoutSeconds: 3600 hardRefreshIntervalSeconds: 3600 pollIntervalSeconds: 30 + # Allow enough time for the backend to prepare the download. + downloadTimeoutSeconds: 3600 + # Allow three hours for the initial import. + progressDeadlineSeconds: 10800 +backendStreamSpool: + directory: /var/lib/loculus/spool + sizeLimit: 20Gi + maxTotalBytes: 19327352832 + maxConcurrentSpools: 4 + minFreeBytes: 1073741824 + fileTtlMinutes: 180 + sweepEveryMinutes: 30 ingestLimitSeconds: 1800 -getSubmissionListLimitSeconds: 600 +# Allow three hours for download and processing. +getSubmissionListLimitSeconds: 10800 preprocessingTimeout: 600 accessionPrefix: "LOC_" dateFieldForGroupGraph: null diff --git a/loculus-silo/README.md b/loculus-silo/README.md index 085db55c6c..a5d3afe020 100644 --- a/loculus-silo/README.md +++ b/loculus-silo/README.md @@ -26,6 +26,7 @@ Environment variables mirror the historical shell scripts: - `HARD_REFRESH_INTERVAL` (seconds, default `3600`) - `SILO_IMPORT_POLL_INTERVAL_SECONDS` (default `30`) - `SILO_RUN_TIMEOUT_SECONDS` (default `3600`) +- `SILO_IMPORT_DOWNLOAD_TIMEOUT_SECONDS` (released data read timeout, default `3600`) - `ROOT_DIR` (optional alternative root for the `/preprocessing` tree) ## Container image diff --git a/loculus-silo/src/silo_import/config.py b/loculus-silo/src/silo_import/config.py index 1c58e75ca0..d13d081378 100644 --- a/loculus-silo/src/silo_import/config.py +++ b/loculus-silo/src/silo_import/config.py @@ -16,6 +16,7 @@ class ImporterConfig: hard_refresh_interval: int poll_interval: int silo_run_timeout: int + download_timeout: int root_dir: Path silo_binary: Path preprocessing_config: Path @@ -56,6 +57,8 @@ def from_env(cls) -> ImporterConfig: hard_refresh_interval = int(env.get("HARD_REFRESH_INTERVAL", "3600")) poll_interval = int(env.get("SILO_IMPORT_POLL_INTERVAL_SECONDS", "30")) silo_run_timeout = int(env.get("SILO_RUN_TIMEOUT_SECONDS", "3600")) + # Large released data downloads can take a long time to start. + download_timeout = int(env.get("SILO_IMPORT_DOWNLOAD_TIMEOUT_SECONDS", "3600")) root_raw = env.get("ROOT_DIR") root_dir = Path(root_raw).resolve() if root_raw else Path("/") silo_binary = Path(env.get("PATH_TO_SILO_BINARY", "/usr/local/bin/silo")) @@ -69,6 +72,7 @@ def from_env(cls) -> ImporterConfig: hard_refresh_interval=hard_refresh_interval, poll_interval=poll_interval, silo_run_timeout=silo_run_timeout, + download_timeout=download_timeout, root_dir=root_dir, silo_binary=silo_binary, preprocessing_config=preprocessing_config, diff --git a/loculus-silo/src/silo_import/download_manager.py b/loculus-silo/src/silo_import/download_manager.py index 93db756a49..f4b11b8077 100644 --- a/loculus-silo/src/silo_import/download_manager.py +++ b/loculus-silo/src/silo_import/download_manager.py @@ -30,6 +30,8 @@ logger = logging.getLogger(__name__) +CONNECT_TIMEOUT_SECONDS = 30 + class RecordCountValidationError(Exception): """Record count does not match expected value.""" @@ -85,7 +87,7 @@ def _download_file( url: str, output_path: Path, etag: str | None = None, - timeout: int = 300, + read_timeout: int = 3600, ) -> HttpResponse: """ Download a file using requests. @@ -94,7 +96,7 @@ def _download_file( url: URL to download from output_path: Where to save the response body etag: Optional ETag for conditional request - timeout: Request timeout in seconds + read_timeout: Read timeout in seconds Returns: HttpResponse with status code and headers @@ -109,7 +111,11 @@ def _download_file( try: # noqa: PLW0717 session = requests.Session() session.headers.update(headers) - response = session.get(url, timeout=timeout, stream=True) + response = session.get( + url, + timeout=(CONNECT_TIMEOUT_SECONDS, read_timeout), + stream=True, + ) # Write response body to file (raw, no automatic decompression) with output_path.open("wb") as f: @@ -174,7 +180,7 @@ def download_release( config.released_data_endpoint, data_path, last_etag, - 300, + config.download_timeout, ) if response.status_code >= BAD_REQUEST: diff --git a/loculus-silo/tests/test_config.py b/loculus-silo/tests/test_config.py index 9ae8f7244e..d1647e6a94 100644 --- a/loculus-silo/tests/test_config.py +++ b/loculus-silo/tests/test_config.py @@ -13,6 +13,8 @@ HARD_REFRESH_INTERVAL = 10 SILO_IMPORT_POLL_INTERVAL_SECONDS = 5 SILO_RUN_TIMEOUT_SECONDS = 99 +SILO_IMPORT_DOWNLOAD_TIMEOUT_SECONDS = 1234 +DEFAULT_DOWNLOAD_TIMEOUT_SECONDS = 3600 def test_config_from_env(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: @@ -23,6 +25,9 @@ def test_config_from_env(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Non monkeypatch.setenv("HARD_REFRESH_INTERVAL", str(HARD_REFRESH_INTERVAL)) monkeypatch.setenv("SILO_IMPORT_POLL_INTERVAL_SECONDS", str(SILO_IMPORT_POLL_INTERVAL_SECONDS)) monkeypatch.setenv("SILO_RUN_TIMEOUT_SECONDS", str(SILO_RUN_TIMEOUT_SECONDS)) + monkeypatch.setenv( + "SILO_IMPORT_DOWNLOAD_TIMEOUT_SECONDS", str(SILO_IMPORT_DOWNLOAD_TIMEOUT_SECONDS) + ) monkeypatch.setenv("ROOT_DIR", str(tmp_path)) config = ImporterConfig.from_env() @@ -33,10 +38,20 @@ def test_config_from_env(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Non assert config.hard_refresh_interval == HARD_REFRESH_INTERVAL assert config.poll_interval == SILO_IMPORT_POLL_INTERVAL_SECONDS assert config.silo_run_timeout == SILO_RUN_TIMEOUT_SECONDS + assert config.download_timeout == SILO_IMPORT_DOWNLOAD_TIMEOUT_SECONDS assert config.root_dir == tmp_path assert config.hierarchical_filters is None +def test_config_download_timeout_defaults(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv("SILO_IMPORT_DOWNLOAD_TIMEOUT_SECONDS", raising=False) + monkeypatch.setenv("BACKEND_BASE_URL", "http://example.com") + + config = ImporterConfig.from_env() + + assert config.download_timeout == DEFAULT_DOWNLOAD_TIMEOUT_SECONDS + + def test_config_missing_backend_env(monkeypatch: pytest.MonkeyPatch) -> None: for key in list(os.environ.keys()): if key.startswith("BACKEND_BASE_URL"): diff --git a/loculus-silo/tests/test_download_manager.py b/loculus-silo/tests/test_download_manager.py new file mode 100644 index 0000000000..89b403e318 --- /dev/null +++ b/loculus-silo/tests/test_download_manager.py @@ -0,0 +1,81 @@ +# ruff: noqa: S101 +from __future__ import annotations + +from pathlib import Path +from unittest.mock import Mock + +import pytest +import requests +from silo_import.config import ImporterConfig +from silo_import.download_manager import ( + CONNECT_TIMEOUT_SECONDS, + DownloadManager, + HttpResponse, + _download_file, # noqa: PLC2701 +) +from silo_import.errors import NotModifiedError +from silo_import.paths import ImporterPaths + +READ_TIMEOUT_SECONDS = 1234 + + +def test_download_file_keeps_connect_timeout_short( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + session = Mock() + session.headers = {} + response = Mock() + response.raw.stream.return_value = [] + response.headers = {} + session.get.return_value = response + monkeypatch.setattr(requests, "Session", lambda: session) + + _download_file( + "http://backend/get-released-data", + tmp_path / "data.zst", + read_timeout=READ_TIMEOUT_SECONDS, + ) + + session.get.assert_called_once_with( + "http://backend/get-released-data", + timeout=(CONNECT_TIMEOUT_SECONDS, READ_TIMEOUT_SECONDS), + stream=True, + ) + + +def test_download_manager_applies_configured_read_timeout(tmp_path: Path) -> None: + seen_timeouts: list[int] = [] + + def download( + _url: str, + output_path: Path, + _etag: str | None, + timeout: int, + ) -> HttpResponse: + seen_timeouts.append(timeout) + output_path.write_bytes(b"") + return HttpResponse(status_code=304, headers={}) + + config = ImporterConfig( + backend_base_url="http://backend", + lineage_definitions=None, + hard_refresh_interval=3600, + poll_interval=30, + silo_run_timeout=3600, + download_timeout=READ_TIMEOUT_SECONDS, + root_dir=tmp_path, + silo_binary=tmp_path / "silo", + preprocessing_config=tmp_path / "config.yaml", + ) + paths = ImporterPaths.from_root( + tmp_path, + config.silo_binary, + config.preprocessing_config, + ) + paths.ensure_directories() + + with pytest.raises(NotModifiedError): + DownloadManager(download_func=download).download_release(config, paths, last_etag="0") + + assert seen_timeouts == [READ_TIMEOUT_SECONDS] diff --git a/loculus-silo/tests/test_integration.py b/loculus-silo/tests/test_integration.py index 482ccec034..a11b87d9af 100644 --- a/loculus-silo/tests/test_integration.py +++ b/loculus-silo/tests/test_integration.py @@ -35,6 +35,7 @@ def make_config( hard_refresh_interval=hard_refresh_interval, poll_interval=1, silo_run_timeout=silo_run_timeout, + download_timeout=5, root_dir=tmp_path, silo_binary=tmp_path / "silo", preprocessing_config=tmp_path / "config.yaml", diff --git a/loculus-silo/tests/test_runner.py b/loculus-silo/tests/test_runner.py index 8153078a89..24ca48cb4e 100644 --- a/loculus-silo/tests/test_runner.py +++ b/loculus-silo/tests/test_runner.py @@ -35,6 +35,7 @@ def make_config( hard_refresh_interval=hard_refresh_interval, poll_interval=1, silo_run_timeout=5, + download_timeout=5, root_dir=tmp_path, silo_binary=tmp_path / "silo", preprocessing_config=tmp_path / "config.yaml",