From e2da479abb37fb07b28ae33ce1397c62e0102913 Mon Sep 17 00:00:00 2001 From: LevinCeglie Date: Thu, 18 Jun 2026 15:04:57 +0200 Subject: [PATCH 01/13] refac(api): overhaul cancel upload pipeline and fix race conditions --- backend/src/services/file.service.ts | 134 +++++++++++++++--- backend/src/services/queue.service.ts | 46 +++--- packages/shared/src/common-utils/enum.ts | 1 + .../file-cleanup-queue-processor.provider.ts | 127 +---------------- 4 files changed, 141 insertions(+), 167 deletions(-) diff --git a/backend/src/services/file.service.ts b/backend/src/services/file.service.ts index c5c931684..fa4c81ed2 100644 --- a/backend/src/services/file.service.ts +++ b/backend/src/services/file.service.ts @@ -19,7 +19,10 @@ import { IngestionJobEntity } from '@kleinkram/backend-common/entities/file/inge import { MissionEntity } from '@kleinkram/backend-common/entities/mission/mission.entity'; import { ProjectEntity } from '@kleinkram/backend-common/entities/project/project.entity'; import env from '@kleinkram/backend-common/environment'; +import { MissionAccessViewEntity } from '@kleinkram/backend-common/viewEntities/mission-access-view.entity'; +import { ProjectAccessViewEntity } from '@kleinkram/backend-common/viewEntities/project-access-view.entity'; import { + AccessGroupRights, DataType, FileEventType, FileOrigin, @@ -44,6 +47,7 @@ import { DataSource, In, MoreThan, + MoreThanOrEqual, QueryFailedError, Repository, SelectQueryBuilder, @@ -1104,6 +1108,7 @@ export class FileService implements OnModuleInit { const disposition = preview_only ? undefined : { + // eslint-disable-next-line @typescript-eslint/naming-convention 'response-content-disposition': `attachment; filename="${file.filename}"`, }; @@ -1368,7 +1373,13 @@ export class FileService implements OnModuleInit { if (fileType === undefined) throw new UnsupportedMediaTypeException(); - if (existingFilenames.has(filename)) { + const existingFile = existingFiles.find( + (f) => f.filename === filename, + ); + const isConflict = + existingFile && existingFile.state !== FileState.CANCELED; + + if (isConflict) { invalidFiles.push({ filename, error: 'File already exists', @@ -1379,19 +1390,32 @@ export class FileService implements OnModuleInit { try { // Use a nested transaction (savepoint) for each file await manager.transaction(async (nestedManager) => { - const file = await nestedManager.save( - FileEntity, - nestedManager.create(FileEntity, { - date: new Date(), - size: 0, - filename, - mission, - creator: user, - type: fileType, - state: FileState.UPLOADING, - origin: FileOrigin.UPLOAD, - }), - ); + let file: FileEntity; + if (existingFile?.state === FileState.CANCELED) { + existingFile.state = FileState.UPLOADING; + existingFile.creator = user; + existingFile.date = new Date(); + existingFile.size = 0; + existingFile.hash = ''; + file = await nestedManager.save( + FileEntity, + existingFile, + ); + } else { + file = await nestedManager.save( + FileEntity, + nestedManager.create(FileEntity, { + date: new Date(), + size: 0, + filename, + mission, + creator: user, + type: fileType, + state: FileState.UPLOADING, + origin: FileOrigin.UPLOAD, + }), + ); + } await this.auditService.log( FileEventType.UPLOAD_STARTED, @@ -1467,12 +1491,84 @@ export class FileService implements OnModuleInit { uuids: string[], missionUUID: string, userUUID: string, - ): Promise { - // Cleanup cannot be done synchronously as this takes too long; the request is sent on unload; delaying the onUnload is difficult and discouraged - return this.fileCleanupQueue.add('cancelUpload', { - uuids, - missionUUID, + ): Promise { + const canCancelUpload = await this.canCancelUpload( userUUID, + missionUUID, + ); + if (!canCancelUpload) { + logger.debug(`User ${userUUID} can't cancel upload`); + return; + } + + await Promise.all( + uuids.map(async (uuid) => { + const file = await this.fileRepository.findOne({ + where: { uuid, mission: { uuid: missionUUID } }, + relations: ['mission'], + }); + if (!file) { + return; + } + if (file.state === FileState.OK) { + return; + } + + if (file.mission === undefined) { + logger.error( + `Mission of file ${file.uuid} is undefined, skipping`, + ); + return; + } + + file.state = FileState.CANCELED; + await this.fileRepository.save(file); + return; + }), + ); + } + + private async canCancelUpload( + userUUID: string, + missionUUID: string, + ): Promise { + const user = await this.userRepository.findOneOrFail({ + where: { uuid: userUUID }, + }); + if (user.role === UserRole.ADMIN) { + return true; + } + const mission = await this.missionRepository.findOneOrFail({ + where: { uuid: missionUUID }, + relations: ['project'], + }); + + if (mission.project === undefined) { + logger.error( + `Project of mission ${mission.uuid} is undefined, skipping`, + ); + return false; + } + + const canAccessProject = await this.dataSource.manager.exists( + ProjectAccessViewEntity, + { + where: { + projectUuid: mission.project.uuid, + userUuid: userUUID, + rights: MoreThanOrEqual(AccessGroupRights.WRITE), + }, + }, + ); + if (canAccessProject) { + return true; + } + return await this.dataSource.manager.exists(MissionAccessViewEntity, { + where: { + missionUuid: missionUUID, + userUuid: userUUID, + rights: MoreThanOrEqual(AccessGroupRights.WRITE), + }, }); } diff --git a/backend/src/services/queue.service.ts b/backend/src/services/queue.service.ts index d74e2037a..5960eedd1 100644 --- a/backend/src/services/queue.service.ts +++ b/backend/src/services/queue.service.ts @@ -154,33 +154,30 @@ export class QueueService implements OnModuleInit { actor: UserEntity, source: FileSource | string = FileSource.WEB_INTERFACE, ): Promise { - let job = await this.queueRepository.findOne({ - where: { identifier: uuid }, + const file = await this.fileRepository.findOneOrFail({ + where: { uuid }, relations: ['mission', 'mission.project'], }); - if (!job) { - logger.warn( - `confirmUpload: Job missing for file ${uuid}.Recreating...`, - ); - - const file = await this.fileRepository.findOneOrFail({ - where: { uuid }, - relations: ['mission', 'mission.project'], - }); + if (file.state === FileState.CANCELED) { + throw new ConflictException('Cannot confirm a canceled upload'); + } - job = await this.queueRepository.save( - this.queueRepository.create({ - identifier: file.uuid, + let job = await this.queueRepository.findOne({ + where: { identifier: uuid }, + relations: ['mission', 'mission.project'], + }); - displayName: file.filename, - state: QueueState.AWAITING_UPLOAD, - location: FileLocation.S3, - mission: file.mission, - creator: actor, - } as IngestionJobEntity), - ); - } + job ??= await this.queueRepository.save( + this.queueRepository.create({ + identifier: file.uuid, + displayName: file.filename, + state: QueueState.AWAITING_UPLOAD, + location: FileLocation.S3, + mission: file.mission, + creator: actor, + } as IngestionJobEntity), + ); if ( job.state !== QueueState.AWAITING_UPLOAD && @@ -193,11 +190,6 @@ export class QueueService implements OnModuleInit { ); } - const file = await this.fileRepository.findOneOrFail({ - where: { uuid: uuid }, - relations: ['mission', 'mission.project'], - }); - const fileInfo = await this.dataStorage .getFileInfo(file.uuid) .catch((error: unknown): void => { diff --git a/packages/shared/src/common-utils/enum.ts b/packages/shared/src/common-utils/enum.ts index 6e2ccfb70..bde85c532 100644 --- a/packages/shared/src/common-utils/enum.ts +++ b/packages/shared/src/common-utils/enum.ts @@ -152,6 +152,7 @@ export enum FileState { CONVERSION_ERROR = 'CONVERSION_ERROR', LOST = 'LOST', FOUND = 'FOUND', + CANCELED = 'CANCELED', } export enum HealthStatus { diff --git a/queueConsumer/src/fileCleanup/file-cleanup-queue-processor.provider.ts b/queueConsumer/src/fileCleanup/file-cleanup-queue-processor.provider.ts index 9e1d8182e..786a957d9 100644 --- a/queueConsumer/src/fileCleanup/file-cleanup-queue-processor.provider.ts +++ b/queueConsumer/src/fileCleanup/file-cleanup-queue-processor.provider.ts @@ -1,40 +1,20 @@ import { redis } from '@kleinkram/backend-common/consts'; import { FileEntity } from '@kleinkram/backend-common/entities/file/file.entity'; import { IngestionJobEntity } from '@kleinkram/backend-common/entities/file/ingestion-job.entity'; -import { MissionEntity } from '@kleinkram/backend-common/entities/mission/mission.entity'; -import { UserEntity } from '@kleinkram/backend-common/entities/user/user.entity'; + import { IStorageBucket } from '@kleinkram/backend-common/modules/storage/types'; -import { MissionAccessViewEntity } from '@kleinkram/backend-common/viewEntities/mission-access-view.entity'; -import { ProjectAccessViewEntity } from '@kleinkram/backend-common/viewEntities/project-access-view.entity'; -import { - AccessGroupRights, - FileState, - QueueState, - UserRole, -} from '@kleinkram/shared'; -import { Process, Processor } from '@nestjs/bull'; +import { FileState, QueueState } from '@kleinkram/shared'; +import { Processor } from '@nestjs/bull'; import { Inject, Injectable, OnModuleInit } from '@nestjs/common'; import { Cron, CronExpression } from '@nestjs/schedule'; import { InjectRepository } from '@nestjs/typeorm'; -import { Job } from 'bull'; + import { Redis } from 'ioredis'; import crypto from 'node:crypto'; import Redlock from 'redlock'; -import { - IsNull, - LessThanOrEqual, - MoreThanOrEqual, - Not, - Repository, -} from 'typeorm'; +import { IsNull, LessThanOrEqual, Not, Repository } from 'typeorm'; import logger from '../logger'; -type CancelUploadJob = Job<{ - uuids: string[]; - missionUUID: string; - userUUID: string; -}>; - @Processor('file-cleanup') @Injectable() export class FileCleanupQueueProcessorProvider implements OnModuleInit { @@ -45,14 +25,7 @@ export class FileCleanupQueueProcessorProvider implements OnModuleInit { private fileRepository: Repository, @InjectRepository(IngestionJobEntity) private queueRepository: Repository, - @InjectRepository(UserEntity) - private userRepository: Repository, - @InjectRepository(MissionEntity) - private missionRepository: Repository, - @InjectRepository(ProjectAccessViewEntity) - private projectAccessView: Repository, - @InjectRepository(MissionAccessViewEntity) - private missionAccessView: Repository, + @Inject('DataStorageBucket') private readonly dataStorage: IStorageBucket, ) {} @@ -65,53 +38,6 @@ export class FileCleanupQueueProcessorProvider implements OnModuleInit { }); } - @Process({ concurrency: 10, name: 'cancelUpload' }) - async process(job: CancelUploadJob): Promise { - const userUUID = job.data.userUUID; - const uuids = job.data.uuids; - const missionUUID = job.data.missionUUID; - const canCancelUpload = await this.canCancelUpload( - userUUID, - missionUUID, - ); - if (!canCancelUpload) { - logger.debug(`User ${userUUID} can't cancel upload`); - return; - } - await Promise.all( - uuids.map(async (uuid) => { - const file = await this.fileRepository.findOne({ - where: { uuid, mission: { uuid: missionUUID } }, - relations: ['mission'], - }); - if (!file) { - return; - } - if (file.state === FileState.OK) { - return; - } - - if (file.mission === undefined) { - logger.error( - `Mission of file ${file.uuid} is undefined, skipping`, - ); - return; - } - - const queue = await this.queueRepository.findOneOrFail({ - where: { - displayName: file.filename, - mission: { uuid: file.mission.uuid }, - }, - }); - queue.state = QueueState.CANCELED; - await this.queueRepository.save(queue); - await this.fileRepository.remove(file); - return; - }), - ); - } - @Cron(CronExpression.EVERY_DAY_AT_3AM) async fixFileHashes(): Promise { await this.redlock @@ -228,45 +154,4 @@ export class FileCleanupQueueProcessorProvider implements OnModuleInit { ); }); } - - async canCancelUpload( - userUUID: string, - missionUUID: string, - ): Promise { - const user = await this.userRepository.findOneOrFail({ - where: { uuid: userUUID }, - }); - if (user.role === UserRole.ADMIN) { - return true; - } - const mission = await this.missionRepository.findOneOrFail({ - where: { uuid: missionUUID }, - relations: ['project'], - }); - - if (mission.project === undefined) { - logger.error( - `Project of mission ${mission.uuid} is undefined, skipping`, - ); - return false; - } - - const canAccessProject = await this.projectAccessView.exists({ - where: { - projectUuid: mission.project.uuid, - userUuid: userUUID, - rights: MoreThanOrEqual(AccessGroupRights.WRITE), - }, - }); - if (canAccessProject) { - return true; - } - return await this.missionAccessView.exists({ - where: { - missionUuid: missionUUID, - userUuid: userUUID, - rights: MoreThanOrEqual(AccessGroupRights.WRITE), - }, - }); - } } From 14640b09b3480d2ff390c1b4f29038a45a7ea117 Mon Sep 17 00:00:00 2001 From: LevinCeglie Date: Thu, 18 Jun 2026 15:35:29 +0200 Subject: [PATCH 02/13] chore(cli+sdk): add new file state to CLI+SDK --- cli/kleinkram/models.py | 1 + cli/kleinkram/printing.py | 1 + 2 files changed, 2 insertions(+) diff --git a/cli/kleinkram/models.py b/cli/kleinkram/models.py index e581da8cd..82489618b 100644 --- a/cli/kleinkram/models.py +++ b/cli/kleinkram/models.py @@ -38,6 +38,7 @@ class FileState(str, Enum): CONVERSION_ERROR = "CONVERSION_ERROR" LOST = "LOST" FOUND = "FOUND" + CANCELED = "CANCELED" @dataclass(frozen=True) diff --git a/cli/kleinkram/printing.py b/cli/kleinkram/printing.py index fb9a4b25f..9f302ee7a 100644 --- a/cli/kleinkram/printing.py +++ b/cli/kleinkram/printing.py @@ -56,6 +56,7 @@ FileState.CONVERSION_ERROR: "red", FileState.LOST: "bold red", FileState.FOUND: "yellow", + FileState.CANCELED: "bright_black", } From b827f40bc2176bd179d83adb20598bde003de774 Mon Sep 17 00:00:00 2001 From: LevinCeglie Date: Tue, 30 Jun 2026 11:13:12 +0200 Subject: [PATCH 03/13] refac(cli+sdk): add retry decorator and refactor file transfor pipeline --- cli/kleinkram/api/file_transfer.py | 57 +++++++----------------------- cli/kleinkram/utils.py | 45 +++++++++++++++++++++++ 2 files changed, 58 insertions(+), 44 deletions(-) diff --git a/cli/kleinkram/api/file_transfer.py b/cli/kleinkram/api/file_transfer.py index 55ea79224..88fdd2374 100644 --- a/cli/kleinkram/api/file_transfer.py +++ b/cli/kleinkram/api/file_transfer.py @@ -29,6 +29,7 @@ from kleinkram.models import FileState from kleinkram.utils import b64_md5 from kleinkram.utils import format_traceback +from kleinkram.utils import retry logger = logging.getLogger(__name__) @@ -61,6 +62,7 @@ class UploadCredentials(NamedTuple): bucket: str +@retry(max_attempts=3, exceptions=(httpx.HTTPError,)) def _confirm_file_upload(client: AuthenticatedClient, file_id: UUID, file_hash: str) -> None: data = { "uuid": str(file_id), @@ -71,6 +73,7 @@ def _confirm_file_upload(client: AuthenticatedClient, file_id: UUID, file_hash: resp.raise_for_status() +@retry(max_attempts=3, exceptions=(httpx.HTTPError,)) def _cancel_file_upload(client: AuthenticatedClient, file_id: UUID, mission_id: UUID) -> None: data = { "uuids": [str(file_id)], @@ -92,28 +95,20 @@ def _cancel_file_upload(client: AuthenticatedClient, file_id: UUID, mission_id: BUCKET_FIELD = "bucket" -def _get_upload_creditials( - client: AuthenticatedClient, internal_filename: str, mission_id: UUID -) -> Optional[UploadCredentials]: +@retry(max_attempts=5, exceptions=(httpx.HTTPError,), exclude_exceptions=(FileExistsError,)) +def _get_upload_creditials(client: AuthenticatedClient, internal_filename: str, mission_id: UUID) -> UploadCredentials: dct = { "filenames": [internal_filename], "missionUUID": str(mission_id), "source": "CLI", } - try: - resp = client.post(UPLOAD_CREDS, json=dct) - resp.raise_for_status() - except httpx.HTTPStatusError as e: - # 409 Conflict means file already exists - if e.response.status_code == 409: - return None - raise + resp = client.post(UPLOAD_CREDS, json=dct) + if resp.status_code == 409: + raise FileExistsError() + resp.raise_for_status() data = resp.json()["data"][0] - if data.get("error") == FILE_EXISTS_ERROR: - return None - bucket = data[BUCKET_FIELD] file_id = UUID(data[FILE_ID_FIELD], version=4) @@ -165,33 +160,6 @@ class UploadState(Enum): CANCELED = 3 -def _get_upload_credentials_with_retry(client, filename, mission_id, max_attempts=5): - """ - Retrieves upload credentials with retry logic. - - Args: - client: The client object used for retrieving credentials. - filename: The internal filename. - mission_id: The mission ID. - max_attempts: Maximum number of retry attempts. - - Returns: - The upload credentials or None if retrieval fails after all attempts. - """ - attempt = 0 - while attempt < max_attempts: - creds = _get_upload_creditials(client, internal_filename=filename, mission_id=mission_id) - if creds is not None: - return creds - - attempt += 1 - if attempt < max_attempts: - delay = 2**attempt # Exponential backoff (2, 4, 8, 16...) - sleep(delay) - - return None - - # TODO: i dont want to handle errors at this level def upload_file( client: AuthenticatedClient, @@ -217,9 +185,9 @@ def upload_file( on_file_start_cb(path, total_size) # get per file upload credentials - creds = _get_upload_credentials_with_retry(client, filename, mission_id, max_attempts=5 if attempt > 0 else 1) - - if creds is None: + try: + creds = _get_upload_creditials(client, internal_filename=filename, mission_id=mission_id) + except FileExistsError: return UploadState.EXISTS, 0 # build the boto3 callback from our file progress callback @@ -239,6 +207,7 @@ def boto3_cb(bytes_amount): _cancel_file_upload(client, creds.file_id, mission_id) except Exception as cancel_e: logger.error(f"Failed to cancel upload for {creds.file_id}: {cancel_e}") + raise RuntimeError(f"Upload failed and cancellation failed for {creds.file_id}: {cancel_e}") from e if attempt < 2: # Retry if not the last attempt logger.warning(f"Retrying upload for {path} (attempt {attempt + 1})") diff --git a/cli/kleinkram/utils.py b/cli/kleinkram/utils.py index 65b40a9dc..cc0068446 100644 --- a/cli/kleinkram/utils.py +++ b/cli/kleinkram/utils.py @@ -1,10 +1,14 @@ from __future__ import annotations import base64 +import functools import hashlib +import logging import math +import random import re import string +import time import traceback from hashlib import md5 from pathlib import Path @@ -27,6 +31,47 @@ from kleinkram.types import IdLike from kleinkram.types import PathLike +logger = logging.getLogger(__name__) + + +def retry( + max_attempts: int = 5, + backoff_base: int = 2, + exceptions: Tuple[type[Exception], ...] = (Exception,), + exclude_exceptions: Tuple[type[Exception], ...] = (), +): + """ + Decorator for retrying a function on specified exceptions with exponential backoff and jitter. + Can override max_attempts dynamically by passing `_retry_attempts` to the wrapped function. + """ + + def decorator(func): + @functools.wraps(func) + def wrapper(*args, **kwargs): + attempts = kwargs.pop("_retry_attempts", max_attempts) + + for attempt in range(1, attempts + 1): + try: + return func(*args, **kwargs) + except exclude_exceptions: + raise + except exceptions as e: + if attempt == attempts: + logger.error(f"Function '{func.__name__}' failed after {attempts} attempts. Final error: {e}") + raise + + delay = (backoff_base**attempt) + random.uniform(0, 0.5) + logger.warning( + f"Function '{func.__name__}' failed (Error: {e}). " + f"Attempt {attempt}/{attempts} failed. Retrying in {delay:.2f}s..." + ) + time.sleep(delay) + + return wrapper + + return decorator + + INTERNAL_ALLOWED_CHARS = string.ascii_letters + string.digits + "_" + "-" SUPPORT_FILE_TYPES = [".bag", ".mcap", ".db3", ".svo2", ".tum", ".yaml", ".yml"] EXPERIMENTAL_FILE_TYPES = [] From 796c21bf3bd73b925b578766e1fe678b0600bba9 Mon Sep 17 00:00:00 2001 From: LevinCeglie Date: Tue, 30 Jun 2026 11:16:43 +0200 Subject: [PATCH 04/13] chore(cli+sdk): add tests for file upload with retries --- cli/tests/test_upload_retry.py | 67 ++++++++++++++++++++++++++++++++++ 1 file changed, 67 insertions(+) create mode 100644 cli/tests/test_upload_retry.py diff --git a/cli/tests/test_upload_retry.py b/cli/tests/test_upload_retry.py new file mode 100644 index 000000000..40d9d7b42 --- /dev/null +++ b/cli/tests/test_upload_retry.py @@ -0,0 +1,67 @@ +from __future__ import annotations + +from pathlib import Path +from unittest.mock import patch + +import pytest + +import kleinkram.api.file_transfer as ft +from kleinkram.api.client import AuthenticatedClient +from kleinkram.api.file_transfer import UploadState +from kleinkram.api.file_transfer import upload_file + + +@pytest.mark.slow +def test_upload_file_retries_on_s3_error(mission, tmp_path): + + # Create a temporary file + test_file = tmp_path / "test_retry.yaml" + test_file.write_text("Hello World for Retry Test") + + client = AuthenticatedClient() + + call_count = {"count": 0} + original_s3_upload = ft._s3_upload + + def mock_s3_upload(*args, **kwargs): + if call_count["count"] == 0: + call_count["count"] += 1 + raise RuntimeError("Simulated S3 Upload Error on first attempt") + else: + call_count["count"] += 1 + return original_s3_upload(*args, **kwargs) + + with patch("kleinkram.api.file_transfer._s3_upload", side_effect=mock_s3_upload): + with patch("kleinkram.api.file_transfer._cancel_file_upload", wraps=ft._cancel_file_upload) as mocked_cancel: + state, size = upload_file(client=client, mission_id=mission.id, filename=test_file.name, path=test_file) + assert state == UploadState.UPLOADED + assert size == len("Hello World for Retry Test") + + assert call_count["count"] == 2 + mocked_cancel.assert_called_once() + + +@pytest.mark.slow +def test_upload_file_aborts_on_cancellation_error(mission, tmp_path): + # Create a temporary file + test_file = tmp_path / "test_retry_abort.yaml" + test_file.write_text("Hello World for Retry Abort Test") + + client = AuthenticatedClient() + + # Simulate S3 upload failure + def mock_s3_upload(*args, **kwargs): + raise RuntimeError("Simulated S3 Upload Error") + + # Simulate cancel upload failure + def mock_cancel_upload(*args, **kwargs): + raise RuntimeError("Simulated Cancellation Error") + + with patch("kleinkram.api.file_transfer._s3_upload", side_effect=mock_s3_upload): + with patch("kleinkram.api.file_transfer._cancel_file_upload", side_effect=mock_cancel_upload) as mocked_cancel: + with pytest.raises(RuntimeError) as exc_info: + upload_file(client=client, mission_id=mission.id, filename=test_file.name, path=test_file) + + # The exception should wrap the cancellation error and abort immediately + assert "cancellation failed" in str(exc_info.value) + mocked_cancel.assert_called_once() From 7b122a4a153c6d0e50ea6bfebfcf2c34342dbec3 Mon Sep 17 00:00:00 2001 From: LevinCeglie Date: Tue, 30 Jun 2026 11:26:47 +0200 Subject: [PATCH 05/13] feat(queueCon): add clean up job for canceled file uploads --- .../file-cleanup-queue-processor.provider.ts | 40 ++++++++++++++++++- 1 file changed, 39 insertions(+), 1 deletion(-) diff --git a/queueConsumer/src/fileCleanup/file-cleanup-queue-processor.provider.ts b/queueConsumer/src/fileCleanup/file-cleanup-queue-processor.provider.ts index 786a957d9..f33f2d558 100644 --- a/queueConsumer/src/fileCleanup/file-cleanup-queue-processor.provider.ts +++ b/queueConsumer/src/fileCleanup/file-cleanup-queue-processor.provider.ts @@ -12,7 +12,7 @@ import { InjectRepository } from '@nestjs/typeorm'; import { Redis } from 'ioredis'; import crypto from 'node:crypto'; import Redlock from 'redlock'; -import { IsNull, LessThanOrEqual, Not, Repository } from 'typeorm'; +import { In, IsNull, LessThanOrEqual, Not, Repository } from 'typeorm'; import logger from '../logger'; @Processor('file-cleanup') @@ -147,6 +147,44 @@ export class FileCleanupQueueProcessorProvider implements OnModuleInit { await this.queueRepository.save(queue); }), ); + + // Clean up canceled uploads older than 24 hours + const canceledUploads = await this.fileRepository.find({ + where: { + state: FileState.CANCELED, + updatedAt: LessThanOrEqual( + new Date(Date.now() - 1000 * 60 * 60 * 24), + ), + }, + }); + if (canceledUploads.length > 0) { + logger.debug( + `Cleaning up ${String(canceledUploads.length)} canceled uploads`, + ); + const canceledUuids = canceledUploads.map((f) => f.uuid); + await this.queueRepository + .softDelete({ + identifier: In(canceledUuids), + }) + .catch((error: unknown) => { + logger.error( + `Failed to soft-delete ingestion jobs for canceled uploads: ${String(error)}`, + ); + }); + + await Promise.all( + canceledUploads.map(async (file) => { + try { + await this.dataStorage.deleteFile(file.uuid); + await this.fileRepository.softRemove(file); + } catch (error: unknown) { + logger.error( + `Failed to clean up canceled upload ${file.uuid}: ${String(error)}`, + ); + } + }), + ); + } }) .catch(() => { logger.debug( From aef3c1048c49b2c0169d4f603f18e1601b4b893f Mon Sep 17 00:00:00 2001 From: LevinCeglie Date: Tue, 30 Jun 2026 14:06:19 +0200 Subject: [PATCH 06/13] fix(sdk+backend): minor fixes --- backend/src/services/file.service.ts | 1 + cli/kleinkram/api/file_transfer.py | 8 ++++---- 2 files changed, 5 insertions(+), 4 deletions(-) diff --git a/backend/src/services/file.service.ts b/backend/src/services/file.service.ts index fa4c81ed2..76b377cfe 100644 --- a/backend/src/services/file.service.ts +++ b/backend/src/services/file.service.ts @@ -1397,6 +1397,7 @@ export class FileService implements OnModuleInit { existingFile.date = new Date(); existingFile.size = 0; existingFile.hash = ''; + existingFile.origin = FileOrigin.UPLOAD; file = await nestedManager.save( FileEntity, existingFile, diff --git a/cli/kleinkram/api/file_transfer.py b/cli/kleinkram/api/file_transfer.py index 88fdd2374..75a2ef810 100644 --- a/cli/kleinkram/api/file_transfer.py +++ b/cli/kleinkram/api/file_transfer.py @@ -62,7 +62,7 @@ class UploadCredentials(NamedTuple): bucket: str -@retry(max_attempts=3, exceptions=(httpx.HTTPError,)) +@retry(max_attempts=3, exceptions=(httpx.TransportError,)) def _confirm_file_upload(client: AuthenticatedClient, file_id: UUID, file_hash: str) -> None: data = { "uuid": str(file_id), @@ -73,7 +73,7 @@ def _confirm_file_upload(client: AuthenticatedClient, file_id: UUID, file_hash: resp.raise_for_status() -@retry(max_attempts=3, exceptions=(httpx.HTTPError,)) +@retry(max_attempts=3, exceptions=(httpx.TransportError,)) def _cancel_file_upload(client: AuthenticatedClient, file_id: UUID, mission_id: UUID) -> None: data = { "uuids": [str(file_id)], @@ -95,7 +95,7 @@ def _cancel_file_upload(client: AuthenticatedClient, file_id: UUID, mission_id: BUCKET_FIELD = "bucket" -@retry(max_attempts=5, exceptions=(httpx.HTTPError,), exclude_exceptions=(FileExistsError,)) +@retry(max_attempts=5, exceptions=(httpx.TransportError,), exclude_exceptions=(FileExistsError,)) def _get_upload_creditials(client: AuthenticatedClient, internal_filename: str, mission_id: UUID) -> UploadCredentials: dct = { "filenames": [internal_filename], @@ -209,7 +209,7 @@ def boto3_cb(bytes_amount): logger.error(f"Failed to cancel upload for {creds.file_id}: {cancel_e}") raise RuntimeError(f"Upload failed and cancellation failed for {creds.file_id}: {cancel_e}") from e - if attempt < 2: # Retry if not the last attempt + if attempt < MAX_UPLOAD_RETRIES - 1: # Retry if not the last attempt logger.warning(f"Retrying upload for {path} (attempt {attempt + 1})") continue else: From e02e30d81ff58330cfb9b137ef7d29b2602545da Mon Sep 17 00:00:00 2001 From: LevinCeglie Date: Tue, 30 Jun 2026 14:17:50 +0200 Subject: [PATCH 07/13] fix(queueCon): minor fix in clean up job for stale canceled uploads --- .../fileCleanup/file-cleanup-queue-processor.provider.ts | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/queueConsumer/src/fileCleanup/file-cleanup-queue-processor.provider.ts b/queueConsumer/src/fileCleanup/file-cleanup-queue-processor.provider.ts index f33f2d558..8e9ca2af1 100644 --- a/queueConsumer/src/fileCleanup/file-cleanup-queue-processor.provider.ts +++ b/queueConsumer/src/fileCleanup/file-cleanup-queue-processor.provider.ts @@ -175,7 +175,13 @@ export class FileCleanupQueueProcessorProvider implements OnModuleInit { await Promise.all( canceledUploads.map(async (file) => { try { - await this.dataStorage.deleteFile(file.uuid); + await this.dataStorage + .deleteFile(file.uuid) + .catch((error: unknown) => { + logger.error( + `Failed to delete S3 object for ${file.uuid}: ${String(error)}`, + ); + }); await this.fileRepository.softRemove(file); } catch (error: unknown) { logger.error( From 0858a8846a11f0909fc241c6962e0bdc307bab95 Mon Sep 17 00:00:00 2001 From: LevinCeglie Date: Tue, 30 Jun 2026 16:29:30 +0200 Subject: [PATCH 08/13] fix(backend): resolve cancelUpload route shadowing bug --- backend/src/endpoints/file/file.controller.ts | 32 +++++++++---------- 1 file changed, 16 insertions(+), 16 deletions(-) diff --git a/backend/src/endpoints/file/file.controller.ts b/backend/src/endpoints/file/file.controller.ts index 8610ec78c..e411c31b1 100644 --- a/backend/src/endpoints/file/file.controller.ts +++ b/backend/src/endpoints/file/file.controller.ts @@ -224,6 +224,22 @@ export class FileController { }); } + @Delete('uploads') + @UserOnly() //Push back authentication to the queue to accelerate the request + @OutputDto(CancelUploadResponseDto) + async cancelUpload( + @Body() dto: CancelFileUploadDto, + @AddUser() auth: AuthHeader, + ): Promise { + logger.debug(`cancelUpload ${JSON.stringify(dto)}`); + await this.fileLifecycleService.cancelUpload( + dto.uuids, + dto.missionUuid, + auth.user.uuid, + ); + return { success: true }; + } + @Delete(':uuid') @CanDeleteFile() @OutputDto(DeleteFileResponseDto) @@ -295,22 +311,6 @@ export class FileController { ); } - @Delete('uploads') - @UserOnly() //Push back authentication to the queue to accelerate the request - @OutputDto(CancelUploadResponseDto) - async cancelUpload( - @Body() dto: CancelFileUploadDto, - @AddUser() auth: AuthHeader, - ): Promise { - logger.debug(`cancelUpload ${JSON.stringify(dto)}`); - await this.fileLifecycleService.cancelUpload( - dto.uuids, - dto.missionUuid, - auth.user.uuid, - ); - return { success: true }; - } - @Delete() @CanDeleteMission() @ApiOkResponse({ From d9b02a8bed6b8685e1660b2bc49fafef4ee6ff4c Mon Sep 17 00:00:00 2001 From: LevinCeglie Date: Tue, 30 Jun 2026 17:13:09 +0200 Subject: [PATCH 09/13] chore(cli+sdk): add exponantial retry backoff to outer most file upload loop --- cli/kleinkram/api/file_transfer.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/cli/kleinkram/api/file_transfer.py b/cli/kleinkram/api/file_transfer.py index 77aaba9e5..35e43f64e 100644 --- a/cli/kleinkram/api/file_transfer.py +++ b/cli/kleinkram/api/file_transfer.py @@ -210,7 +210,8 @@ def boto3_cb(bytes_amount): raise RuntimeError(f"Upload failed and cancellation failed for {creds.file_id}: {cancel_e}") from e if attempt < MAX_UPLOAD_RETRIES - 1: # Retry if not the last attempt - logger.warning(f"Retrying upload for {path} (attempt {attempt + 1})") + logger.warning(f"Retrying upload for {path} (attempt {attempt + 1}), retrying after backoff...") + sleep(RETRY_BACKOFF_BASE**attempt) continue else: logger.error(f"Cancelling upload for {path} after {attempt + 1} attempts") From e840d4695899dfd1fe503fcc9326877fb2f7e17e Mon Sep 17 00:00:00 2001 From: LevinCeglie Date: Wed, 1 Jul 2026 10:24:36 +0200 Subject: [PATCH 10/13] fix(queueCon): minor fix to clean up job of canceled uploads --- .../src/fileCleanup/file-cleanup-queue-processor.provider.ts | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/queueConsumer/src/fileCleanup/file-cleanup-queue-processor.provider.ts b/queueConsumer/src/fileCleanup/file-cleanup-queue-processor.provider.ts index 0ff949b92..3200d3cd8 100644 --- a/queueConsumer/src/fileCleanup/file-cleanup-queue-processor.provider.ts +++ b/queueConsumer/src/fileCleanup/file-cleanup-queue-processor.provider.ts @@ -186,7 +186,10 @@ export class FileCleanupQueueProcessorProvider implements OnModuleInit { `Failed to delete S3 object for ${file.uuid}: ${String(error)}`, ); }); - await this.fileRepository.softRemove(file); + await this.fileRepository.softDelete({ + uuid: file.uuid, + state: FileState.CANCELED, + }); } catch (error: unknown) { logger.error( `Failed to clean up canceled upload ${file.uuid}: ${String(error)}`, From 04bae7a113a9a20aef914a94551757a1907bfbe8 Mon Sep 17 00:00:00 2001 From: LevinCeglie Date: Wed, 1 Jul 2026 13:27:24 +0200 Subject: [PATCH 11/13] feat(backend+frondent+CLI/SDK): add pre-flight check to /files/temporaryAccess for checking storage capacitity of SeaweedFS bucket --- backend/src/endpoints/file/file.controller.ts | 1 + .../src/services/file-lifecycle.service.ts | 22 ++ backend/tests/files/upload-preflight.test.ts | 212 ++++++++++++++++++ cli/kleinkram/api/file_transfer.py | 17 +- cli/kleinkram/cli/error_handling.py | 28 +++ cli/kleinkram/errors.py | 3 + cli/tests/test_upload_retry.py | 23 ++ frontend/src/services/file-service.ts | 33 ++- .../file/temporary-access-request.dto.ts | 13 ++ 9 files changed, 340 insertions(+), 12 deletions(-) create mode 100644 backend/tests/files/upload-preflight.test.ts diff --git a/backend/src/endpoints/file/file.controller.ts b/backend/src/endpoints/file/file.controller.ts index e411c31b1..7cde5e3e0 100644 --- a/backend/src/endpoints/file/file.controller.ts +++ b/backend/src/endpoints/file/file.controller.ts @@ -308,6 +308,7 @@ export class FileController { auth.user.uuid, auth.apiKey?.action, source, + body.fileSizes, ); } diff --git a/backend/src/services/file-lifecycle.service.ts b/backend/src/services/file-lifecycle.service.ts index d53a6a4ae..e46614c13 100644 --- a/backend/src/services/file-lifecycle.service.ts +++ b/backend/src/services/file-lifecycle.service.ts @@ -27,6 +27,8 @@ import { import { BadRequestException, ConflictException, + HttpException, + HttpStatus, Inject, Injectable, NotFoundException, @@ -334,6 +336,7 @@ export class FileLifecycleService implements OnModuleInit { userUUID: string, action?: ActionEntity, uploadSource = 'Web Interface', + fileSizes?: number[], ): Promise { const mission = await this.missionRepository.findOneOrFail({ where: { uuid: missionUUID }, @@ -343,6 +346,25 @@ export class FileLifecycleService implements OnModuleInit { where: { uuid: userUUID }, }); + if ( + fileSizes && + fileSizes.length > 0 && + this.dataStorage.getSystemMetrics + ) { + const metrics = await this.dataStorage.getSystemMetrics(); + const freeBytes = metrics.totalBytes - metrics.usedBytes; + const totalRequestedSize = fileSizes.reduce( + (sum, size) => sum + size, + 0, + ); + if (totalRequestedSize > freeBytes) { + throw new HttpException( + 'Insufficient storage space on the server', + HttpStatus.INSUFFICIENT_STORAGE, + ); + } + } + return await this.dataSource.transaction(async (manager) => { // Deduplicate filenames to avoid self-collisions const uniqueFilenames = [...new Set(filenames)]; diff --git a/backend/tests/files/upload-preflight.test.ts b/backend/tests/files/upload-preflight.test.ts new file mode 100644 index 000000000..932cd4fc5 --- /dev/null +++ b/backend/tests/files/upload-preflight.test.ts @@ -0,0 +1,212 @@ +jest.mock('@kleinkram/backend-common/environment', () => ({ + /* eslint-disable-next-line @typescript-eslint/naming-convention */ + __esModule: true, + default: { + S3_DATA_BUCKET_NAME: 'test-bucket', + }, +})); + +import { + CategoryEntity, + FileEntity, + MissionEntity, + UserEntity, +} from '@kleinkram/backend-common'; +import { FileAuditService } from '@kleinkram/backend-common/audit/file-audit.service'; +import { IStorageBucket } from '@kleinkram/backend-common/modules/storage/types'; +import { HttpException, HttpStatus } from '@nestjs/common'; +import { DataSource, Repository } from 'typeorm'; +import { FileLifecycleService } from '../../src/services/file-lifecycle.service'; +import { TriggerService } from '../../src/services/trigger.service'; + +describe('File Upload Pre-Flight Capacity Check', () => { + let service: FileLifecycleService; + let mockGetSystemMetrics: jest.Mock; + let mockMissionRepo: jest.Mocked>; + let mockUserRepo: jest.Mocked>; + let mockDataSource: jest.Mocked; + + beforeEach(() => { + mockGetSystemMetrics = jest.fn().mockResolvedValue({ + usedBytes: 800, + totalBytes: 1000, // 200 free bytes + }); + + mockMissionRepo = { + findOneOrFail: jest.fn().mockResolvedValue({ + uuid: 'mission-uuid', + project: { uuid: 'project-uuid' }, + }), + } as unknown as jest.Mocked>; + + mockUserRepo = { + findOneOrFail: jest.fn().mockResolvedValue({ uuid: 'user-uuid' }), + } as unknown as jest.Mocked>; + + const mockDataStorage = { + getSystemMetrics: mockGetSystemMetrics, + generateTemporaryCredential: jest.fn().mockResolvedValue({}), + } as unknown as jest.Mocked; + + mockDataSource = { + transaction: jest + .fn() + .mockImplementation( + (callback: (mgr: unknown) => Promise) => + callback({ + find: jest.fn().mockResolvedValue([]), + transaction: jest + .fn() + .mockImplementation( + ( + callback2: ( + mgr2: unknown, + ) => Promise, + ) => + callback2({ + save: jest.fn().mockResolvedValue({ + uuid: 'file-uuid', + filename: 'test.bag', + }), + create: jest + .fn() + .mockReturnValue({}), + }), + ), + }), + ), + } as unknown as jest.Mocked; + + service = new FileLifecycleService( + {} as unknown as jest.Mocked>, + mockMissionRepo, + mockUserRepo, + {} as unknown as jest.Mocked>, + mockDataStorage, + mockDataSource, + { log: jest.fn() } as unknown as jest.Mocked, + { + addFileEvent: jest.fn(), + } as unknown as jest.Mocked, + ); + }); + + it('should generate credentials if requested size fits within free space', async () => { + const result = await service.getTemporaryAccess( + ['test.bag'], + 'mission-uuid', + 'user-uuid', + undefined, + 'CLI', + [150], // 150 bytes requested < 200 free bytes + ); + expect(result).toBeDefined(); + expect(mockGetSystemMetrics).toHaveBeenCalledTimes(1); + }); + + it('should allow upload when requested size exactly equals free space', async () => { + const result = await service.getTemporaryAccess( + ['test.bag'], + 'mission-uuid', + 'user-uuid', + undefined, + 'CLI', + [200], // exactly 200 free bytes + ); + expect(result).toBeDefined(); + expect(mockGetSystemMetrics).toHaveBeenCalledTimes(1); + }); + + it('should throw INSUFFICIENT_STORAGE if requested size exceeds free space', async () => { + await expect( + service.getTemporaryAccess( + ['test.bag'], + 'mission-uuid', + 'user-uuid', + undefined, + 'CLI', + [250], // 250 bytes requested > 200 free bytes + ), + ).rejects.toThrow( + new HttpException( + 'Insufficient storage space on the server', + HttpStatus.INSUFFICIENT_STORAGE, + ), + ); + }); + + it('should sum all file sizes for multi-file uploads', async () => { + // Two files: 100 + 150 = 250 > 200 free bytes + await expect( + service.getTemporaryAccess( + ['a.bag', 'b.bag'], + 'mission-uuid', + 'user-uuid', + undefined, + 'CLI', + [100, 150], + ), + ).rejects.toThrow( + new HttpException( + 'Insufficient storage space on the server', + HttpStatus.INSUFFICIENT_STORAGE, + ), + ); + }); + + it('should skip capacity check when fileSizes is not provided', async () => { + const result = await service.getTemporaryAccess( + ['test.bag'], + 'mission-uuid', + 'user-uuid', + undefined, + 'CLI', + // no fileSizes argument + ); + expect(result).toBeDefined(); + expect(mockGetSystemMetrics).not.toHaveBeenCalled(); + }); + + it('should skip capacity check when fileSizes is an empty array', async () => { + const result = await service.getTemporaryAccess( + ['test.bag'], + 'mission-uuid', + 'user-uuid', + undefined, + 'CLI', + [], + ); + expect(result).toBeDefined(); + expect(mockGetSystemMetrics).not.toHaveBeenCalled(); + }); + + it('should skip capacity check if getSystemMetrics is not implemented on the storage bucket', async () => { + // Create a custom service instance where the storage bucket does not have getSystemMetrics + const mockDataStorageNoMetrics = { + generateTemporaryCredential: jest.fn().mockResolvedValue({}), + } as unknown as jest.Mocked; + + const serviceNoMetrics = new FileLifecycleService( + {} as unknown as jest.Mocked>, + mockMissionRepo, // reuse the mocked repo + mockUserRepo, // reuse the mocked repo + {} as unknown as jest.Mocked>, + mockDataStorageNoMetrics, + mockDataSource, // reuse the mocked dataSource + { log: jest.fn() } as unknown as jest.Mocked, + { + addFileEvent: jest.fn(), + } as unknown as jest.Mocked, + ); + + const result = await serviceNoMetrics.getTemporaryAccess( + ['test.bag'], + 'mission-uuid', + 'user-uuid', + undefined, + 'CLI', + [150], + ); + expect(result).toBeDefined(); + }); +}); diff --git a/cli/kleinkram/api/file_transfer.py b/cli/kleinkram/api/file_transfer.py index 35e43f64e..a808df92a 100644 --- a/cli/kleinkram/api/file_transfer.py +++ b/cli/kleinkram/api/file_transfer.py @@ -25,6 +25,7 @@ from kleinkram.api.client import AuthenticatedClient from kleinkram.config import get_config from kleinkram.errors import AccessDenied +from kleinkram.errors import InsufficientStorageError from kleinkram.models import File from kleinkram.models import FileState from kleinkram.utils import b64_md5 @@ -95,16 +96,21 @@ def _cancel_file_upload(client: AuthenticatedClient, file_id: UUID, mission_id: BUCKET_FIELD = "bucket" -@retry(max_attempts=5, exceptions=(httpx.TransportError,), exclude_exceptions=(FileExistsError,)) -def _get_upload_creditials(client: AuthenticatedClient, internal_filename: str, mission_id: UUID) -> UploadCredentials: +@retry(max_attempts=5, exceptions=(httpx.TransportError,), exclude_exceptions=(FileExistsError, InsufficientStorageError)) +def _get_upload_creditials( + client: AuthenticatedClient, internal_filename: str, mission_id: UUID, file_size: int +) -> UploadCredentials: dct = { "filenames": [internal_filename], "missionUUID": str(mission_id), "source": "CLI", + "fileSizes": [file_size], } resp = client.post(UPLOAD_CREDS, json=dct) if resp.status_code == 409: raise FileExistsError() + if resp.status_code == 507: + raise InsufficientStorageError("Insufficient storage space on the server") resp.raise_for_status() data = resp.json()["data"][0] @@ -186,7 +192,7 @@ def upload_file( # get per file upload credentials try: - creds = _get_upload_creditials(client, internal_filename=filename, mission_id=mission_id) + creds = _get_upload_creditials(client, internal_filename=filename, mission_id=mission_id, file_size=total_size) except FileExistsError: return UploadState.EXISTS, 0 @@ -498,6 +504,11 @@ def upload_files( try: state, size_bytes = future.result() + except InsufficientStorageError as e: + if on_message_cb is not None: + on_message_cb("Upload failed: Insufficient storage space on the server", True) + executor.shutdown(wait=False, cancel_futures=True) + raise e except Exception as e: logger.error(format_traceback(e)) if on_message_cb is not None: diff --git a/cli/kleinkram/cli/error_handling.py b/cli/kleinkram/cli/error_handling.py index 77eb2749c..4d03aeae8 100644 --- a/cli/kleinkram/cli/error_handling.py +++ b/cli/kleinkram/cli/error_handling.py @@ -16,6 +16,7 @@ from kleinkram.config import get_config from kleinkram.config import get_shared_state +from kleinkram.errors import InsufficientStorageError from kleinkram.utils import format_traceback from kleinkram.utils import upper_camel_case_to_words @@ -197,11 +198,38 @@ def handle_http_status_error(exc: httpx.HTTPStatusError) -> int: return 1 +def handle_insufficient_storage_error(exc: InsufficientStorageError) -> int: + shared_state = get_shared_state() + config = get_config() + endpoint_url = config.endpoint.api + + title = "Insufficient Storage Space" + msg = ( + f"The Kleinkram backend server at:\n" + f" [bold cyan]{endpoint_url}[/bold cyan] is out of storage space.\n\n" + f"The requested file upload(s) exceed the remaining capacity of the S3 bucket." + ) + quiet_msg = "Error: Insufficient storage space on the server" + + display_error( + exc=exc, + verbose=shared_state.verbose, + title=title, + message=msg if shared_state.verbose else quiet_msg, + ) + logger.error(f"Insufficient storage error on {endpoint_url}: {exc}") + + if shared_state.debug: + raise exc + return 1 + + def register_error_handlers(app: ErrorHandledTyper) -> None: """Register all CLI error handlers. Note: since ErrorHandledTyper dispatches handlers in reverse order of registration, the most generic Exception handler must be registered first, and more specific handlers (like httpx.RequestError) later. """ app.error_handler(Exception)(handle_generic_exception) + app.error_handler(InsufficientStorageError)(handle_insufficient_storage_error) app.error_handler(httpx.RequestError)(handle_request_error) app.error_handler(httpx.HTTPStatusError)(handle_http_status_error) diff --git a/cli/kleinkram/errors.py b/cli/kleinkram/errors.py index 8a2cfe262..922541f39 100644 --- a/cli/kleinkram/errors.py +++ b/cli/kleinkram/errors.py @@ -87,3 +87,6 @@ class TriggerValidationError(Exception): ... class TriggerNotFound(Exception): ... + + +class InsufficientStorageError(Exception): ... diff --git a/cli/tests/test_upload_retry.py b/cli/tests/test_upload_retry.py index 40d9d7b42..b8e214083 100644 --- a/cli/tests/test_upload_retry.py +++ b/cli/tests/test_upload_retry.py @@ -65,3 +65,26 @@ def mock_cancel_upload(*args, **kwargs): # The exception should wrap the cancellation error and abort immediately assert "cancellation failed" in str(exc_info.value) mocked_cancel.assert_called_once() + + +@pytest.mark.slow +def test_upload_file_fails_on_insufficient_storage(mission, tmp_path): + from httpx import Response + + from kleinkram.errors import InsufficientStorageError + + # Create a temporary file + test_file = tmp_path / "test_insufficient_storage.yaml" + test_file.write_text("Hello World for Storage Test") + + client = AuthenticatedClient() + + # Mock client.post to return 507 when requesting credentials + mock_response = Response(status_code=507) + + with patch.object(client, "post", return_value=mock_response) as mock_post: + with pytest.raises(InsufficientStorageError) as exc_info: + upload_file(client=client, mission_id=mission.id, filename=test_file.name, path=test_file) + + assert "Insufficient storage space on the server" in str(exc_info.value) + mock_post.assert_called_once() diff --git a/frontend/src/services/file-service.ts b/frontend/src/services/file-service.ts index 453957f7e..b6ebf151a 100644 --- a/frontend/src/services/file-service.ts +++ b/frontend/src/services/file-service.ts @@ -222,10 +222,12 @@ async function _createFileAction( while (!temporaryCredentials) { const filenames = fileItems.map((item) => item.virtualName); + const fileSizes = fileItems.map((item) => item.file.size); try { temporaryCredentials = await generateTemporaryCredentials({ filenames, missionUUID: selectedMission.uuid, + fileSizes, }); } catch (error: unknown) { let handled = false; @@ -305,15 +307,28 @@ async function _createFileAction( let message = `Upload failed: ${errorMessage}`; if (error instanceof AxiosError) { - if (error.response?.status === 403) { - message = `Upload failed: You do not have the necessary permissions.`; - } else if (error.response?.status === 400) { - const responseData: unknown = error.response.data; - message = - isValidationErrorResponse(responseData) && - responseData.message - ? `Upload failed: ${responseData.message}` - : `Upload failed: ${JSON.stringify(responseData)}`; + switch (error.response?.status) { + case 403: { + message = `Upload failed: You do not have the necessary permissions.`; + + break; + } + case 507: { + message = `Upload failed: Insufficient storage space on the server. Please free up space before trying again.`; + + break; + } + case 400: { + const responseData: unknown = error.response.data; + message = + isValidationErrorResponse(responseData) && + responseData.message + ? `Upload failed: ${responseData.message}` + : `Upload failed: ${JSON.stringify(responseData)}`; + + break; + } + // No default } } diff --git a/packages/api-dto/src/types/file/temporary-access-request.dto.ts b/packages/api-dto/src/types/file/temporary-access-request.dto.ts index 9cb396644..1560976b7 100644 --- a/packages/api-dto/src/types/file/temporary-access-request.dto.ts +++ b/packages/api-dto/src/types/file/temporary-access-request.dto.ts @@ -2,8 +2,10 @@ import { FileSource } from '@kleinkram/shared'; import { IsNoValidUUID, IsValidFileName } from '@kleinkram/validation'; import { ApiProperty } from '@nestjs/swagger'; import { + IsArray, IsEnum, IsNotEmpty, + IsNumber, IsOptional, IsString, IsUUID, @@ -34,4 +36,15 @@ export class TemporaryAccessRequestDto { enum: FileSource, }) source?: FileSource; + + @IsNumber({}, { each: true }) + @IsArray() + @IsOptional() + @ApiProperty({ + description: + 'Sizes of the files in bytes matching the order of filenames', + required: false, + type: [Number], + }) + fileSizes?: number[]; } From 2741926e19d683ae33c03460579fb0e6dde61dbd Mon Sep 17 00:00:00 2001 From: LevinCeglie Date: Wed, 1 Jul 2026 13:28:48 +0200 Subject: [PATCH 12/13] fix(CLI/SDK): fix minor issue in deserializer regarding file categories --- cli/kleinkram/api/deser.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/cli/kleinkram/api/deser.py b/cli/kleinkram/api/deser.py index 219ec6f4a..c5bfd8601 100644 --- a/cli/kleinkram/api/deser.py +++ b/cli/kleinkram/api/deser.py @@ -226,7 +226,8 @@ def _parse_file(file: FileObject) -> File: created_at = _parse_datetime(file[FileObjectKeys.CREATED_AT]) updated_at = _parse_datetime(file[FileObjectKeys.UPDATED_AT]) state = _parse_file_state(file[FileObjectKeys.STATE]) - categories = file[FileObjectKeys.CATEGORIES] + categories_raw = file.get(FileObjectKeys.CATEGORIES) or [] + categories = [c["name"] if isinstance(c, dict) and "name" in c else str(c) for c in categories_raw] mission_id, mission_name = _get_nested_info(file, MISSION) project_id, project_name = _get_nested_info(file[MISSION], PROJECT) From c458441660ba2cf14b5d4a57e53e3f4e89fd92e9 Mon Sep 17 00:00:00 2001 From: LevinCeglie Date: Wed, 1 Jul 2026 13:37:45 +0200 Subject: [PATCH 13/13] chore(CLI/SDK): add error handling for SeaweedFS insufficient storage error --- cli/kleinkram/api/file_transfer.py | 45 ++++++++++++++++++++++++++++++ cli/tests/test_upload_retry.py | 36 ++++++++++++++++++++++++ 2 files changed, 81 insertions(+) diff --git a/cli/kleinkram/api/file_transfer.py b/cli/kleinkram/api/file_transfer.py index a808df92a..371931f4d 100644 --- a/cli/kleinkram/api/file_transfer.py +++ b/cli/kleinkram/api/file_transfer.py @@ -21,6 +21,7 @@ import boto3.s3.transfer import botocore.config import httpx +from botocore.exceptions import ClientError from kleinkram.api.client import AuthenticatedClient from kleinkram.config import get_config @@ -160,6 +161,42 @@ def _s3_upload( ) +def _is_s3_out_of_space_error(e: Exception) -> bool: + if isinstance(e, ClientError): + response = e.response + error = response.get("Error", {}) + code = error.get("Code", "") + message = error.get("Message", "") + status_code = response.get("ResponseMetadata", {}).get("HTTPStatusCode", 0) + + if status_code == 507: + return True + + out_of_space_codes = { + "InsufficientStorageSpace", + "QuotaExceeded", + "StorageLimitExceeded", + "QuotaExceededException", + } + if code in out_of_space_codes: + return True + + message_lower = message.lower() + if "insufficient storage" in message_lower or "out of space" in message_lower or "no space left" in message_lower: + return True + + e_str = str(e).lower() + if ( + "insufficient storage" in e_str + or "out of space" in e_str + or "no space left" in e_str + or "507 insufficient storage" in e_str + ): + return True + + return False + + class UploadState(Enum): UPLOADED = 1 EXISTS = 2 @@ -208,6 +245,14 @@ def boto3_cb(bytes_amount): try: _s3_upload(path, endpoint=s3_endpoint, credentials=creds, callback=boto3_cb) except Exception as e: + if _is_s3_out_of_space_error(e): + logger.error("Upload failed: S3 storage is out of space.") + try: + _cancel_file_upload(client, creds.file_id, mission_id) + except Exception as cancel_e: + logger.error(f"Failed to cancel upload for {creds.file_id}: {cancel_e}") + raise InsufficientStorageError("Insufficient storage space on the server") from e + logger.error(format_traceback(e)) try: _cancel_file_upload(client, creds.file_id, mission_id) diff --git a/cli/tests/test_upload_retry.py b/cli/tests/test_upload_retry.py index b8e214083..50f749803 100644 --- a/cli/tests/test_upload_retry.py +++ b/cli/tests/test_upload_retry.py @@ -88,3 +88,39 @@ def test_upload_file_fails_on_insufficient_storage(mission, tmp_path): assert "Insufficient storage space on the server" in str(exc_info.value) mock_post.assert_called_once() + + +@pytest.mark.slow +def test_upload_file_fails_mid_upload_on_s3_out_of_space(mission, tmp_path): + from botocore.exceptions import ClientError + + from kleinkram.errors import InsufficientStorageError + + # Create a temporary file + test_file = tmp_path / "test_s3_out_of_space.yaml" + test_file.write_text("Hello World for S3 Out of Space Test") + + client = AuthenticatedClient() + + # Create a simulated ClientError for out of space + error_response = { + "Error": { + "Code": "InsufficientStorageSpace", + "Message": "There is not enough space on the S3 device.", + }, + "ResponseMetadata": { + "HTTPStatusCode": 507, + }, + } + simulated_s3_error = ClientError(error_response, "PutObject") + + def mock_s3_upload(*args, **kwargs): + raise simulated_s3_error + + with patch("kleinkram.api.file_transfer._s3_upload", side_effect=mock_s3_upload): + with patch("kleinkram.api.file_transfer._cancel_file_upload", wraps=ft._cancel_file_upload) as mocked_cancel: + with pytest.raises(InsufficientStorageError) as exc_info: + upload_file(client=client, mission_id=mission.id, filename=test_file.name, path=test_file) + + assert "Insufficient storage space on the server" in str(exc_info.value) + mocked_cancel.assert_called_once()