Merge pull request #1414 from leggedrobotics/dev

Release 0.43.4

Author: wp99cp

backend/package.json

@@ -1,6 +1,6 @@
 {
     "name": "kleinkram-backend",
-    "version": "0.43.3",
+    "version": "0.43.4",
     "description": "",
     "author": "",
     "private": true,

backend/src/endpoints/file/file.controller.ts

@@ -54,6 +54,7 @@ import {
 } from '@common/api/types/file/access.dto';
 
 import { FileQueryDto } from '@common/api/types/file/file-query.dto';
+import { CancelFileUploadDto } from '@common/api/types/cancel-file-upload.dto';
 
 @Controller(['file', 'files']) // TODO: migrate to 'files'
 export class FileController {
@@ -286,20 +287,15 @@ export class FileController {
 
     @Post('cancelUpload')
     @UserOnly() //Push back authentication to the queue to accelerate the request
-    // TODO: type API response
+    @OutputDto(null) // TODO: type API response
     async cancelUpload(
-        @BodyUUIDArray(
-            'uuids',
-            "File UUIDs who, if they aren't 'OK', are deleted",
-        )
-        uuids: string[],
-        @BodyUUID('missionUUID', 'Mission UUID') missionUUID: string,
+        @Body() dto: CancelFileUploadDto,
         @AddUser() auth: AuthHeader,
-    ) {
-        logger.debug(`cancelUpload ${uuids.toString()}`);
-        return this.fileService.cancelUpload(
-            uuids,
-            missionUUID,
+    ): Promise<void> {
+        logger.debug(`cancelUpload ${dto.uuids.toString()}`);
+        await this.fileService.cancelUpload(
+            dto.uuids,
+            dto.missionUuid,
             auth.user.uuid,
         );
     }

backend/src/routing/middlewares/version-checker-middleware.service.ts

@@ -49,9 +49,7 @@ export class VersionCheckerMiddlewareService implements NestMiddleware {
         }
 
         // forbidden client versions: allows for the following notations
-        // - '<0.40.x': versions below 0.40.x are forbidden
-        // - '0.41.x': version 0.41.x is forbidden
-        const forbiddenClientVersions = ['<0.40.0', '0.41.0', '0.41.1'];
+        const forbiddenClientVersions = ['<0.43.4'];
 
         if (this.isVersionForbidden(clientVersion, forbiddenClientVersions)) {
             this.rejectRequest(response, clientVersion);

cli/kleinkram/api/file_transfer.py

@@ -1,13 +1,14 @@
 from __future__ import annotations
 
+import signal
 import logging
 import sys
 from concurrent.futures import Future
 from concurrent.futures import ThreadPoolExecutor
 from concurrent.futures import as_completed
 from enum import Enum
 from pathlib import Path
-from time import monotonic
+from time import monotonic, sleep
 from typing import Dict
 from typing import NamedTuple
 from typing import Optional
@@ -39,6 +40,7 @@
 DOWNLOAD_CHUNK_SIZE = 1024 * 1024 * 16
 DOWNLOAD_URL = "/files/download"
 
+MAX_UPLOAD_RETRIES = 3
 S3_MAX_RETRIES = 60  # same as frontend
 S3_READ_TIMEOUT = 60 * 5  # 5 minutes
 
@@ -66,8 +68,8 @@ def _cancel_file_upload(
     client: AuthenticatedClient, file_id: UUID, mission_id: UUID
 ) -> None:
     data = {
-        "uuid": [str(file_id)],
-        "missionUUID": str(mission_id),
+        "uuids": [str(file_id)],
+        "missionUuid": str(mission_id),
     }
     resp = client.post(UPLOAD_CANCEL, json=data)
     resp.raise_for_status()
@@ -151,6 +153,37 @@ class UploadState(Enum):
     CANCELED = 3
 
 
+def _get_upload_credentials_with_retry(
+    client, pbar, 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,
@@ -161,40 +194,54 @@ def upload_file(
     verbose: bool = False,
     s3_endpoint: Optional[str] = None,
 ) -> Tuple[UploadState, int]:
-    """\
+    """
     returns UploadState and bytes uploaded (0 if not uploaded)
+    Retries up to 3 times on failure.
     """
     if s3_endpoint is None:
         s3_endpoint = get_config().endpoint.s3
 
     total_size = path.stat().st_size
-    with tqdm(
-        total=total_size,
-        unit="B",
-        unit_scale=True,
-        desc=f"uploading {path}...",
-        leave=False,
-        disable=not verbose,
-    ) as pbar:
-        # get per file upload credentials
-        creds = _get_upload_creditials(
-            client, internal_filename=filename, mission_id=mission_id
-        )
-        if creds is None:
-            return UploadState.EXISTS, 0
+    for attempt in range(MAX_UPLOAD_RETRIES):
+        with tqdm(
+            total=total_size,
+            unit="B",
+            unit_scale=True,
+            desc=f"uploading {path}...",
+            leave=False,
+            disable=not verbose,
+        ) as pbar:
+
+            # get per file upload credentials
+            creds = _get_upload_credentials_with_retry(
+                client, pbar, filename, mission_id, max_attempts=5 if attempt > 0 else 1
+            )
+
+            if creds is None:
+                return UploadState.EXISTS, 0
 
-        try:
-            _s3_upload(path, endpoint=s3_endpoint, credentials=creds, pbar=pbar)
-        except Exception as e:
-            logger.error(format_traceback(e))
             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 e from e
-        else:
-            _confirm_file_upload(client, creds.file_id, b64_md5(path))
-            return UploadState.UPLOADED, total_size
+                _s3_upload(path, endpoint=s3_endpoint, credentials=creds, pbar=pbar)
+            except Exception as e:
+                logger.error(format_traceback(e))
+                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}"
+                    )
+
+                if attempt < 2:  # Retry if not the last attempt
+                    pbar.update(0)
+                    logger.error(f"Retrying upload for {attempt + 1}")
+                    continue
+                else:
+                    logger.error(f"Cancelling upload for {attempt}")
+                    raise e from e
+
+            else:
+                _confirm_file_upload(client, creds.file_id, b64_md5(path))
+                return UploadState.UPLOADED, total_size
 
 
 def _get_file_download(client: AuthenticatedClient, id: UUID) -> str:
@@ -420,6 +467,9 @@ def upload_files(
     ) as pbar:
         start = monotonic()
         futures: Dict[Future[Tuple[UploadState, int]], Path] = {}
+
+        skipped_files = 0
+        failed_files = 0
         with ThreadPoolExecutor(max_workers=n_workers) as executor:
             for name, path in files.items():
                 if not path.is_file():
@@ -441,6 +491,16 @@ def upload_files(
 
             total_uploaded_bytes = 0
             for future in as_completed(futures):
+
+                if future.exception():
+                    failed_files += 1
+
+                if (
+                    future.exception() is None
+                    and future.result()[0] == UploadState.EXISTS
+                ):
+                    skipped_files += 1
+
                 path = futures[future]
                 uploaded_bytes = _upload_handler(future, path, verbose=verbose)
                 total_uploaded_bytes += uploaded_bytes
@@ -455,6 +515,16 @@ def upload_files(
         console.print(f"Total uploaded: {format_bytes(total_uploaded_bytes)}")
         console.print(f"Average speed: {format_bytes(avg_speed_bps, speed=True)}")
 
+        if failed_files > 0:
+            console.print(
+                f"\nUploaded {len(files) - failed_files - skipped_files} files, {skipped_files} skipped, {failed_files} uploads failed",
+                style="red",
+            )
+        else:
+            console.print(
+                f"\nUploaded {len(files) - skipped_files} files, {skipped_files} skipped"
+            )
+
 
 def download_files(
     client: AuthenticatedClient,

cli/setup.cfg

@@ -1,6 +1,6 @@
 [metadata]
 name = kleinkram
-version = 0.43.3
+version = 0.43.4
 description = give me your bags
 long_description = file: README.md
 long_description_content_type = text/markdown

common/api/types/cancel-file-upload.dto.ts

@@ -0,0 +1,12 @@
+import { IsArray, IsString, IsUUID } from 'class-validator';
+import { ApiProperty } from '@nestjs/swagger';
+
+export class CancelFileUploadDto {
+    @ApiProperty()
+    @IsArray()
+    @IsString({ each: true })
+    uuids!: string[];
+
+    @IsUUID()
+    missionUuid!: string;
+}

common/package.json

@@ -1,6 +1,6 @@
 {
     "name": "kleinkram-common",
-    "version": "0.43.3",
+    "version": "0.43.4",
     "license": "MIT",
     "scripts": {
         "seed:config": "ts-node ./node_modules/typeorm-seeding/dist/cli.js config",

docs/package.json

@@ -1,6 +1,6 @@
 {
     "name": "kleinkram-docs",
-    "version": "0.43.3",
+    "version": "0.43.4",
     "license": "MIT",
     "target": "es2022",
     "scripts": {

frontend/package.json

@@ -1,6 +1,6 @@
 {
     "name": "kleinkram-frontend",
-    "version": "0.43.3",
+    "version": "0.43.4",
     "description": "Data storage of ROS bags",
     "productName": "Kleinkram",
     "author": "Johann Schwabe <[email protected]>",

frontend/src/services/mutations/file.ts

@@ -52,12 +52,12 @@ export const generateTemporaryCredentials = async (
 };
 
 export const cancelUploads = async (
-    fileUUIDs: string[],
-    missionUUID: string,
-) => {
+    fileUuids: string[],
+    missionUuid: string,
+): Promise<void> => {
     const response = await axios.post('/file/cancelUpload', {
-        uuids: fileUUIDs,
-        missionUUID,
+        uuids: fileUuids,
+        missionUuid: missionUuid,
     });
     return response.data;
 };