add more logs

Author: Pulkit0110

google/cloud/storage/_experimental/asyncio/async_appendable_object_writer.py

@@ -174,7 +174,7 @@ def __init__(
             overwriting existing objects).
 
             Warning: If `None`, a new object is created. If an object with the
-            same name already exists, it will be overwritten the moment 
+            same name already exists, it will be overwritten the moment
             `writer.open()` is called.
 
         :type write_handle: bytes
@@ -432,8 +432,8 @@ async def generator():
                             print(f"AsyncAppendableObjectWriter: append - sending state_lookup request: {chunk_req}")
                             await self.write_obj_stream.send(chunk_req)
                             resp = await self.write_obj_stream.recv()
-                            print(f"AsyncAppendableObjectWriter: append - received state_lookup response: {resp}")
-                            
+                            print(f"AsyncAppendableObjectWriter: append - received state_lookup response")
+
                             # Update state from open response
                             if resp:
                                 if resp.persisted_size is not None:
@@ -444,15 +444,15 @@ async def generator():
                                     self.write_handle = resp.write_handle
                                     write_state.write_handle = resp.write_handle
                             continue
-                        
+
                         # This is a data request - send it
-                        print(f"AsyncAppendableObjectWriter: append - sending data request: {chunk_req}")
+                        print(f"AsyncAppendableObjectWriter: append - sending data request:")
                         await self.write_obj_stream.send(chunk_req)
 
                     # Get final response from the last request (which has state_lookup=True)
                     resp = await self.write_obj_stream.recv()
-                    print(f"AsyncAppendableObjectWriter: append - received final response: {resp}")
-                    
+                    print(f"AsyncAppendableObjectWriter: append - received final response")
+
                     if resp:
                         if resp.persisted_size is not None:
                             self.persisted_size = resp.persisted_size

google/cloud/storage/_experimental/asyncio/async_write_object_stream.py

@@ -169,7 +169,7 @@ async def open(self, metadata: Optional[List[Tuple[str, str]]] = None) -> None:
         await self.socket_like_rpc.open()  # this is actually 1 send
         print(f"in _AsyncWriteObjectStream open: rpc opened")
         response = await self.socket_like_rpc.recv()
-        print(f"in _AsyncWriteObjectStream open: received response: {response}")
+        print(f"in _AsyncWriteObjectStream open: received response:")
         self._is_stream_open = True
 
         if response.persisted_size:
@@ -212,7 +212,7 @@ async def send(
                 The request message to send. This is typically used to specify
                 the read offset and limit.
         """
-        print(f"in _AsyncWriteObjectStream send: sending request: {bidi_write_object_request}")
+        print(f"in _AsyncWriteObjectStream send: sending request")
         if not self._is_stream_open:
             raise ValueError("Stream is not open")
         await self.socket_like_rpc.send(bidi_write_object_request)
@@ -231,7 +231,7 @@ async def recv(self) -> _storage_v2.BidiWriteObjectResponse:
             raise ValueError("Stream is not open")
         print(f"in _AsyncWriteObjectStream recv: receiving response")
         response = await self.socket_like_rpc.recv()
-        print(f"in _AsyncWriteObjectStream recv: received response: {response}")
+        print(f"in _AsyncWriteObjectStream recv: received response")
         # Update write_handle if present in response
         if response:
             if response.write_handle:

google/cloud/storage/_experimental/asyncio/retry/writes_resumption_strategy.py

@@ -54,6 +54,7 @@ def __init__(
         )
         self.chunk_size = chunk_size
         self.user_buffer = user_buffer
+        self.total_size = self.user_buffer.getbuffer().nbytes
         self.persisted_size: int = 0
         self.bytes_sent: int = 0
         self.bytes_since_last_flush: int = 0
@@ -109,11 +110,10 @@ def generate_requests(self, state: Dict[str, Any]):
             if not chunk:
                 break
 
-            # Peek to see if this is the last chunk. This is safe because both
-            # io.BytesIO and BufferedReader (used in file uploads) support peek().
-            is_last_chunk = not getattr(write_state.user_buffer, "peek", lambda n: b"")(
-                1
-            )
+            chunk_len = len(chunk)
+            is_last_chunk = (
+                write_state.bytes_sent + chunk_len
+            ) == write_state.total_size
 
             checksummed_data = storage_type.ChecksummedData(content=chunk)
             checksum = google_crc32c.Checksum(chunk)
@@ -123,24 +123,25 @@ def generate_requests(self, state: Dict[str, Any]):
                 write_offset=write_state.bytes_sent,
                 checksummed_data=checksummed_data,
             )
-            chunk_len = len(chunk)
+
             write_state.bytes_sent += chunk_len
             write_state.bytes_since_last_flush += chunk_len
             print(f"Yielding request with offset: {request.write_offset}")
 
-            if (
+            is_flush_point = (
                 write_state.flush_interval
                 and write_state.bytes_since_last_flush >= write_state.flush_interval
-            ):
-                request.flush = True
-                print("Marking request with flush=True")
-                # reset counter after marking flush
-                write_state.bytes_since_last_flush = 0
+            )
 
             if is_last_chunk:
                 request.flush = True
                 request.state_lookup = True
+                write_state.bytes_since_last_flush = 0
                 print("Marking request with flush=True and state_lookup=True")
+            elif is_flush_point:
+                request.flush = True
+                write_state.bytes_since_last_flush = 0
+                print("Marking request with flush=True")
 
             yield request