google-cloud-storage: AsyncAppendableObjectWriter.append() silently drops data after a stream reconnect on non-first appends

[bvaradar]

Summary

AsyncAppendableObjectWriter.append() can silently drop data after a stream reconnect when the append is not the first one written to the object. The retry path repositions the per-call buffer using the object-wide persisted_size, so the unpersisted remainder of the current append() call is never resent. finalize() (without full_object_checksum) then completes the object short, with no error.

Environment

• google-cloud-storage 3.15.0 (same code on main of google-cloud-python today)
• Python 3.12, zonal (Rapid) bucket, AsyncAppendableObjectWriter over AsyncGrpcClient

Where

• packages/google-cloud-storage/google/cloud/storage/asyncio/async_appendable_object_writer.py, append(): each call wraps only its own bytes in a new buffer (buffer = io.BytesIO(data)), while write_state.bytes_sent = self.offset and write_state.persisted_size = self.persisted_size are object-wide offsets.
• packages/google-cloud-storage/google/cloud/storage/asyncio/retry/writes_resumption_strategy.py, recover_state_on_failure(), and the re-open branch in append(): write_state.user_buffer.seek(write_state.persisted_size).

For the first append() (offset 0) the two coordinate systems coincide, so it works. For any later append(), persisted_size is past the end of the per-call buffer: read() returns nothing, no bytes are resent, bytes_sent jumps to persisted_size, and the next append() continues from there. The missing range is the part of the current call that the server had not persisted (at most one append()'s worth).

Minimal reproduction (no network; uses the library's own strategy code)

import asyncio, io
from google.cloud.storage.asyncio.retry.writes_resumption_strategy import (
    _WriteState, _WriteResumptionStrategy)

PIECE = 1000
strategy = _WriteResumptionStrategy()

# Third append() call: object already holds 2000 bytes. As in append():
# fresh buffer with this call's bytes; object-wide bytes_sent / persisted_size.
buf = io.BytesIO(b"C" * PIECE)
ws = _WriteState(chunk_size=250, user_buffer=buf, flush_interval=10**9)
ws.bytes_sent = ws.persisted_size = 2000
state = {"write_state": ws}

sent = strategy.generate_requests(state)[:2]   # 500 bytes of this call sent
ws.bytes_sent, _ = 2500, buf.seek(500)

ws.persisted_size = 2250                       # stream breaks; server persisted 250 of them
asyncio.run(strategy.recover_state_on_failure(RuntimeError("stream reset"), state))
resent = sum(len(r.checksummed_data.content) for r in strategy.generate_requests(state))
print("buffer position:", buf.tell(), "of", PIECE)   # 2250 of 1000
print("bytes resent:", resent, "owed:", PIECE - 250)  # 0 owed 750

Output:

buffer position: 2250 of 1000
bytes resent: 0 owed: 750

Observed in practice

Writing ~1.3 GB objects to a zonal bucket with append() in 64 MiB pieces (FLUSH_INTERVAL_BYTES 64 MiB) followed by close(finalize_on_close=True), 1 of ~2,500 objects was finalized 60 MiB short with no exception; reads returned the short object. The missing amount is less than one append piece, consistent with the path above. (Client INFO logs were not captured for that run, so the reconnect itself was not directly observed.)

Expected

After a reconnect, the buffer should be positioned relative to the start of the current append() call (e.g. seek(persisted_size - start_offset_of_this_call)), or data sent since the last confirmed persist should be retained so it can be resent. If the gap cannot be recovered, append() should raise rather than advance offset.

Workaround

Pass full_object_checksum to close(finalize_on_close=True) / finalize() so the server rejects a mismatched object, or compare the total bytes appended with persisted_size before finalizing.