From 42afdd3dc61efaa8762a3ef16e50eb1bed03e344 Mon Sep 17 00:00:00 2001 From: nidhiii-27 Date: Tue, 31 Mar 2026 15:19:00 +0530 Subject: [PATCH 1/3] fix: fix a race condition caused due to concurrent writes --- gcs/upload.py | 12 +++++++----- 1 file changed, 7 insertions(+), 5 deletions(-) diff --git a/gcs/upload.py b/gcs/upload.py index b1985e8c..ec48279d 100644 --- a/gcs/upload.py +++ b/gcs/upload.py @@ -595,9 +595,10 @@ def update_upload_checksums(upload_metadata, object_checksums): update_upload_checksums(upload.metadata, object_checksums) def update_appendable_blob(blob, unused_generation): - blob.media = upload.media - blob.metadata.size = len(upload.media) - blob.metadata.checksums.crc32c = crc32c.crc32c(upload.media) + snapshot = bytes(upload.media) + blob.media = snapshot + blob.metadata.size = snapshot + blob.metadata.checksums.crc32c = crc32c.crc32c(snapshot) return blob blob = db.do_update_object( @@ -632,11 +633,12 @@ def update_appendable_blob(blob, unused_generation): if is_appendable: def finalize_blob(blob, unused_generation): - blob.media = upload.media + snapshot = bytes(upload.media) + blob.media = snapshot blob.metadata.finalize_time.FromDatetime( datetime.datetime.now(datetime.timezone.utc) ) - blob.metadata.size = len(upload.media) + blob.metadata.size = len(snapshot) blob.upload = None blob.upload_gen = 0 return blob From a01297c3f4399ae4ae84763d6f303e9c2fa3f219 Mon Sep 17 00:00:00 2001 From: nidhiii-27 Date: Tue, 31 Mar 2026 16:11:24 +0530 Subject: [PATCH 2/3] fix: add thread locking --- gcs/upload.py | 13 +++++++++---- 1 file changed, 9 insertions(+), 4 deletions(-) diff --git a/gcs/upload.py b/gcs/upload.py index ec48279d..8fe525b7 100644 --- a/gcs/upload.py +++ b/gcs/upload.py @@ -31,6 +31,7 @@ import gcs import testbench +import threading from google.storage.v2 import storage_pb2 @@ -58,6 +59,7 @@ def init(cls, request, metadata, bucket, location, upload_id): media=b"", complete=False, transfer=set(), + lock=threading.Lock(), ) @classmethod @@ -585,7 +587,8 @@ def update_upload_checksums(upload_metadata, object_checksums): # Currently, the testbench will always checkpoint and flush data for testing purposes, # instead of the 15 seconds interval used in the GCS server. # TODO(#592): Refactor testbench checkpointing to more closely follow GCS server behavior. - upload.media += content + with upload.lock: + upload.media += content # Update appendable blob size and media here, as part of #720. # TODO(#720): (a) Update crc and update_time in object metadata. @@ -595,9 +598,10 @@ def update_upload_checksums(upload_metadata, object_checksums): update_upload_checksums(upload.metadata, object_checksums) def update_appendable_blob(blob, unused_generation): - snapshot = bytes(upload.media) + with upload.lock: + snapshot = bytes(upload.media) blob.media = snapshot - blob.metadata.size = snapshot + blob.metadata.size = len(snapshot) blob.metadata.checksums.crc32c = crc32c.crc32c(snapshot) return blob @@ -633,7 +637,8 @@ def update_appendable_blob(blob, unused_generation): if is_appendable: def finalize_blob(blob, unused_generation): - snapshot = bytes(upload.media) + with upload.lock: + snapshot = bytes(upload.media) blob.media = snapshot blob.metadata.finalize_time.FromDatetime( datetime.datetime.now(datetime.timezone.utc) From ac285732915758b30abe0906eaed1dd09c1ea335 Mon Sep 17 00:00:00 2001 From: nidhiii-27 Date: Tue, 31 Mar 2026 16:23:13 +0530 Subject: [PATCH 3/3] fix lint --- gcs/upload.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/gcs/upload.py b/gcs/upload.py index 8fe525b7..7bc5025b 100644 --- a/gcs/upload.py +++ b/gcs/upload.py @@ -18,6 +18,7 @@ import hashlib import itertools import json +import threading import types import uuid @@ -31,7 +32,6 @@ import gcs import testbench -import threading from google.storage.v2 import storage_pb2