fix: prevent data loss on multipart complete and part retry

- Serialize metadata save before state deletion in complete_multipart
  to prevent DEK loss if metadata save fails
- Subtract old part size before adding new on part retry to prevent
  total_plaintext_size from doubling

Author: ServerSideHannes

s3proxy/handlers/multipart/lifecycle.py

@@ -160,21 +160,22 @@ async def handle_complete_multipart_upload(
                     e, client, bucket, key, upload_id, s3_parts, completed_parts, total_plaintext
                 )
 
-            # Save metadata and cleanup
+            # Save metadata first, then delete state.
+            # Order matters: if metadata save fails, state is preserved
+            # so the upload can be retried. Deleting state first would
+            # lose the DEK, making the object permanently undecryptable.
             wrapped_dek = crypto.wrap_key(state.dek, self.settings.kek)
-            await asyncio.gather(
-                save_multipart_metadata(
-                    client, bucket, key,
-                    MultipartMetadata(
-                        version=1,
-                        part_count=len(completed_parts),
-                        total_plaintext_size=total_plaintext,
-                        parts=completed_parts,
-                        wrapped_dek=wrapped_dek,
-                    ),
+            await save_multipart_metadata(
+                client, bucket, key,
+                MultipartMetadata(
+                    version=1,
+                    part_count=len(completed_parts),
+                    total_plaintext_size=total_plaintext,
+                    parts=completed_parts,
+                    wrapped_dek=wrapped_dek,
                 ),
-                delete_upload_state(client, bucket, key, upload_id),
             )
+            await delete_upload_state(client, bucket, key, upload_id)
 
             logger.info(
                 "COMPLETE_MULTIPART_SUCCESS",

s3proxy/state/manager.py

@@ -152,6 +152,9 @@ def updater(data: bytes) -> bytes:
             if state is None:
                 raise StateMissingError(f"Upload state corrupted for {bucket}/{key}")
 
+            old_part = state.parts.get(part.part_number)
+            if old_part is not None:
+                state.total_plaintext_size -= old_part.plaintext_size
             state.parts[part.part_number] = part
             state.total_plaintext_size += part.plaintext_size
             if max_internal >= state.next_internal_part_number: