From 34628c5653b1289835e745ab19e025c204e7fa8f Mon Sep 17 00:00:00 2001 From: serversidehannes Date: Sun, 12 Jul 2026 18:06:21 +0200 Subject: [PATCH 1/2] fix: report plaintext size/etag for multipart objects in listings MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ListObjects resolved plaintext size from the backend object's user metadata, which is only stamped on the simple PUT path. Multipart objects carry their plaintext size in the .meta sidecar (user metadata is fixed at CreateMultipartUpload), so listings reported the ciphertext size. Sync clients comparing sizes (scylla-manager's rclone) saw every multipart object as changed and re-uploaded it on every pass — doubling backup transfer and defeating snapshot dedup entirely. _process_list_objects now falls back to the sidecar when the head metadata lacks plaintext-size, returning the same synthetic etag that HEAD and CompleteMultipartUpload report. Resolved attributes are kept in a process-local LRU keyed by (bucket, key, backend etag) — an overwrite changes the backend etag, which invalidates the entry — and CompleteMultipartUpload primes the cache so fresh uploads list without extra round-trips. --- s3proxy/handlers/buckets.py | 51 +++- s3proxy/handlers/multipart/lifecycle.py | 14 +- s3proxy/state/__init__.py | 7 + s3proxy/state/attr_cache.py | 60 ++++ .../test_list_multipart_plaintext_size.py | 271 ++++++++++++++++++ tests/unit/test_list_objects_parallel.py | 1 + 6 files changed, 392 insertions(+), 12 deletions(-) create mode 100644 s3proxy/state/attr_cache.py create mode 100644 tests/unit/test_list_multipart_plaintext_size.py diff --git a/s3proxy/handlers/buckets.py b/s3proxy/handlers/buckets.py index 4961a30..c2c8436 100644 --- a/s3proxy/handlers/buckets.py +++ b/s3proxy/handlers/buckets.py @@ -15,7 +15,14 @@ from .. import xml_responses from ..client import S3Credentials from ..errors import S3Error -from ..state import INTERNAL_PREFIX, META_SUFFIX_LEGACY, delete_multipart_metadata +from ..state import ( + INTERNAL_PREFIX, + META_SUFFIX_LEGACY, + delete_multipart_metadata, + load_multipart_metadata, + plaintext_attr_cache, + synthetic_multipart_etag, +) from ..xml_utils import find_element, find_elements from .base import BaseHandler @@ -211,24 +218,46 @@ async def _process_list_objects(self, client, bucket: str, contents: list[dict]) sem = asyncio.Semaphore(LIST_HEAD_CONCURRENCY) async def resolve(obj: dict) -> dict: + backend_etag = str(obj.get("ETag", "")).strip('"') + cached = plaintext_attr_cache.get(bucket, obj["Key"], backend_etag) + if cached is not None: + size, etag = cached + return self._list_entry(obj, size, etag) async with sem: try: head = await client.head_object(bucket, obj["Key"]) meta = head.get("Metadata", {}) - size = self._get_plaintext_size(meta, obj.get("Size", 0)) - etag = self._get_effective_etag(meta, obj.get("ETag", "")) + if "plaintext-size" in meta: + size = self._get_plaintext_size(meta, obj.get("Size", 0)) + etag = self._get_effective_etag(meta, obj.get("ETag", "")) + elif mp_meta := await load_multipart_metadata(client, bucket, obj["Key"]): + # Multipart objects can't carry plaintext-size in user + # metadata (it is fixed at CreateMultipartUpload); the + # size lives in the .meta sidecar. Reporting the backend + # Size here would leak the ciphertext size and make sync + # clients re-upload every multipart object on each pass. + size = mp_meta.total_plaintext_size + etag = synthetic_multipart_etag(size) + else: + size = self._get_plaintext_size(meta, obj.get("Size", 0)) + etag = self._get_effective_etag(meta, obj.get("ETag", "")) + plaintext_attr_cache.put(bucket, obj["Key"], backend_etag, size, etag) except Exception: - size, etag = obj.get("Size", 0), obj.get("ETag", "").strip('"') - return { - "key": obj["Key"], - "last_modified": _s3_timestamp(obj["LastModified"]), - "etag": etag, - "size": size, - "storage_class": obj.get("StorageClass", "STANDARD"), - } + size, etag = obj.get("Size", 0), backend_etag + return self._list_entry(obj, size, etag) return await asyncio.gather(*[resolve(obj) for obj in listable]) + @staticmethod + def _list_entry(obj: dict, size: int, etag: str) -> dict: + return { + "key": obj["Key"], + "last_modified": _s3_timestamp(obj["LastModified"]), + "etag": etag, + "size": size, + "storage_class": obj.get("StorageClass", "STANDARD"), + } + async def handle_create_bucket(self, request: Request, creds: S3Credentials) -> Response: bucket = self._parse_bucket(request.url.path) async with self._client(creds) as client: diff --git a/s3proxy/handlers/multipart/lifecycle.py b/s3proxy/handlers/multipart/lifecycle.py index 2aab7ab..f7af824 100644 --- a/s3proxy/handlers/multipart/lifecycle.py +++ b/s3proxy/handlers/multipart/lifecycle.py @@ -23,7 +23,9 @@ PartMetadata, delete_upload_state, persist_upload_state, + plaintext_attr_cache, save_multipart_metadata, + synthetic_multipart_etag, ) from ...xml_utils import find_elements, get_element_text from ..base import BaseHandler @@ -179,11 +181,21 @@ async def handle_complete_multipart_upload( # Complete in S3 try: - await client.complete_multipart_upload(bucket, key, upload_id, s3_parts) + complete_resp = await client.complete_multipart_upload( + bucket, key, upload_id, s3_parts + ) except ClientError as e: await self._handle_complete_error( e, client, bucket, key, upload_id, s3_parts, completed_parts, total_plaintext ) + else: + plaintext_attr_cache.put( + bucket, + key, + str(complete_resp.get("ETag", "")).strip('"'), + total_plaintext, + synthetic_multipart_etag(total_plaintext), + ) # Save metadata first, then delete state. # Order matters: if metadata save fails, state is preserved diff --git a/s3proxy/state/__init__.py b/s3proxy/state/__init__.py index cb84bb2..2ad3520 100644 --- a/s3proxy/state/__init__.py +++ b/s3proxy/state/__init__.py @@ -8,6 +8,9 @@ - Recovery logic for lost Redis state """ +# Plaintext attribute cache for listings +from .attr_cache import PlaintextAttrCache, plaintext_attr_cache, synthetic_multipart_etag + # State manager and storage from .manager import MAX_INTERNAL_PARTS_PER_CLIENT, MultipartStateManager @@ -81,6 +84,10 @@ "load_upload_state", "persist_upload_state", "save_multipart_metadata", + # Plaintext attribute cache + "PlaintextAttrCache", + "plaintext_attr_cache", + "synthetic_multipart_etag", # Recovery "reconstruct_upload_state_from_s3", # Serialization diff --git a/s3proxy/state/attr_cache.py b/s3proxy/state/attr_cache.py new file mode 100644 index 0000000..99c1eb9 --- /dev/null +++ b/s3proxy/state/attr_cache.py @@ -0,0 +1,60 @@ +"""Process-local cache of plaintext attributes for completed objects. + +Multipart objects can't carry their plaintext size in backend user metadata +(user metadata is fixed at CreateMultipartUpload, before the final size is +known), so list pages must resolve it from the .meta sidecar. This cache +remembers the resolved (size, etag) per (bucket, key, backend_etag) so +repeated listings — e.g. a backup client walking the same directories on +every sync pass — skip the per-object round-trips. Keying by backend ETag +scopes an entry to one object version: overwriting the key changes the +backend ETag, which invalidates the cached attributes without coordination. +""" + +import hashlib +from collections import OrderedDict + +# ~100k entries of (bucket, key, etag) -> (size, etag) stays in the tens of +# MB even with long backup keys — well inside the pod memory limit. +_DEFAULT_MAXSIZE = 100_000 + + +def synthetic_multipart_etag(plaintext_size: int) -> str: + """ETag reported for multipart-encrypted objects. + + Must stay in sync with what CompleteMultipartUpload and HEAD return for + these objects, so listings, HEADs, and upload responses all agree. + """ + return hashlib.md5(str(plaintext_size).encode(), usedforsecurity=False).hexdigest() + + +class PlaintextAttrCache: + def __init__(self, maxsize: int = _DEFAULT_MAXSIZE) -> None: + self._maxsize = maxsize + self._entries: OrderedDict[tuple[str, str, str], tuple[int, str]] = OrderedDict() + + def get(self, bucket: str, key: str, backend_etag: str) -> tuple[int, str] | None: + if not backend_etag: + return None + entry_key = (bucket, key, backend_etag) + entry = self._entries.get(entry_key) + if entry is not None: + self._entries.move_to_end(entry_key) + return entry + + def put(self, bucket: str, key: str, backend_etag: str, size: int, etag: str) -> None: + if not backend_etag: + return + entry_key = (bucket, key, backend_etag) + self._entries[entry_key] = (size, etag) + self._entries.move_to_end(entry_key) + while len(self._entries) > self._maxsize: + self._entries.popitem(last=False) + + def clear(self) -> None: + self._entries.clear() + + def __len__(self) -> int: + return len(self._entries) + + +plaintext_attr_cache = PlaintextAttrCache() diff --git a/tests/unit/test_list_multipart_plaintext_size.py b/tests/unit/test_list_multipart_plaintext_size.py new file mode 100644 index 0000000..19fe598 --- /dev/null +++ b/tests/unit/test_list_multipart_plaintext_size.py @@ -0,0 +1,271 @@ +"""LIST must report plaintext size/etag for multipart-encrypted objects. + +Multipart uploads can't stamp ``plaintext-size`` user metadata on the backend +object (user metadata is fixed at CreateMultipartUpload, before the final size +is known) — their plaintext size lives in the ``.meta`` sidecar. If +``_process_list_objects`` falls back to the raw backend Size it reports the +ciphertext size, and sync clients (scylla-manager's rclone) see "sizes differ" +and re-upload every multipart object on every pass. + +These tests pin: +- sidecar resolution for multipart objects (size + synthetic etag, matching + what HEAD and CompleteMultipartUpload report), +- raw-size fallback when there is no sidecar (legacy/unencrypted object), +- user-metadata resolution still winning for simple PUT objects, +- the attr cache eliminating repeat backend round-trips, scoped to the + backend ETag so overwrites re-resolve, +- CompleteMultipartUpload priming the cache. +""" + +import asyncio +import datetime as dt +from urllib.parse import urlencode + +import pytest + +from s3proxy import crypto +from s3proxy.handlers.buckets import BucketHandlerMixin +from s3proxy.state import ( + MultipartMetadata, + encode_multipart_metadata, + plaintext_attr_cache, + synthetic_multipart_etag, +) +from s3proxy.state.attr_cache import PlaintextAttrCache +from s3proxy.state.metadata import _internal_meta_key, persist_upload_state + +INTERNAL_PREFIX = ".s3proxy-internal/" + + +@pytest.fixture(autouse=True) +def fresh_cache(monkeypatch): + """Isolate each test from the process-global cache singleton.""" + cache = PlaintextAttrCache() + monkeypatch.setattr("s3proxy.handlers.buckets.plaintext_attr_cache", cache) + plaintext_attr_cache.clear() + yield cache + plaintext_attr_cache.clear() + + +class _Body: + def __init__(self, data: bytes) -> None: + self._data = data + + async def read(self) -> bytes: + return self._data + + +class FakeHandler: + _process_list_objects = BucketHandlerMixin._process_list_objects + _list_entry = staticmethod(BucketHandlerMixin._list_entry) + + def _is_internal_key(self, key): + return key.startswith(INTERNAL_PREFIX) + + def _get_plaintext_size(self, meta, fallback): + return int(meta.get("plaintext-size", fallback)) + + def _get_effective_etag(self, meta, fallback): + return meta.get("client-etag", fallback.strip('"')) + + +class FakeClient: + """Backend with per-key user metadata and .meta sidecars.""" + + def __init__(self, metadata=None, sidecars=None, fail_head_key=None): + self.metadata = metadata or {} + self.sidecars = sidecars or {} + self.fail_head_key = fail_head_key + self.head_calls = 0 + self.get_calls = 0 + + async def head_object(self, bucket, key): + self.head_calls += 1 + if key == self.fail_head_key: + raise RuntimeError("backend HEAD failed") + return {"Metadata": self.metadata.get(key, {})} + + async def get_object(self, bucket, key): + self.get_calls += 1 + if key in self.sidecars: + encoded = encode_multipart_metadata(self.sidecars[key]) + return {"Body": _Body(encoded.encode())} + raise KeyError(key) + + +def _obj(key, size, etag="backend-etag"): + return { + "Key": key, + "Size": size, + "ETag": f'"{etag}"', + "LastModified": dt.datetime(2026, 7, 12, 9, 0, 0), + "StorageClass": "STANDARD", + } + + +def _mp_meta(plaintext_size): + return MultipartMetadata( + version=2, part_count=3, total_plaintext_size=plaintext_size, wrapped_dek=b"x" * 40 + ) + + +def test_multipart_size_resolved_from_sidecar(): + handler = FakeHandler() + client = FakeClient(sidecars={_internal_meta_key("mp.db"): _mp_meta(12345)}) + + result = asyncio.run( + handler._process_list_objects(client, "bucket", [_obj("mp.db", size=99999)]) + ) + + assert result[0]["size"] == 12345 + assert result[0]["etag"] == synthetic_multipart_etag(12345) + + +def test_no_sidecar_falls_back_to_listed_size(): + handler = FakeHandler() + client = FakeClient() + + result = asyncio.run( + handler._process_list_objects(client, "bucket", [_obj("legacy.bin", size=777)]) + ) + + assert result[0]["size"] == 777 + assert result[0]["etag"] == "backend-etag" + + +def test_simple_put_metadata_wins_without_sidecar_lookup(): + handler = FakeHandler() + client = FakeClient( + metadata={"small.txt": {"plaintext-size": "42", "client-etag": "abc"}}, + ) + + result = asyncio.run( + handler._process_list_objects(client, "bucket", [_obj("small.txt", size=100)]) + ) + + assert result[0]["size"] == 42 + assert result[0]["etag"] == "abc" + assert client.get_calls == 0 + + +def test_repeat_listing_served_from_cache(): + handler = FakeHandler() + client = FakeClient( + metadata={"small.txt": {"plaintext-size": "42", "client-etag": "abc"}}, + sidecars={_internal_meta_key("mp.db"): _mp_meta(12345)}, + ) + contents = [_obj("mp.db", size=99999), _obj("small.txt", size=100)] + + first = asyncio.run(handler._process_list_objects(client, "bucket", contents)) + head_calls, get_calls = client.head_calls, client.get_calls + second = asyncio.run(handler._process_list_objects(client, "bucket", contents)) + + assert first == second + assert client.head_calls == head_calls + assert client.get_calls == get_calls + + +def test_cache_scoped_to_backend_etag(): + handler = FakeHandler() + client = FakeClient(sidecars={_internal_meta_key("mp.db"): _mp_meta(12345)}) + + asyncio.run(handler._process_list_objects(client, "bucket", [_obj("mp.db", 99999, "v1")])) + head_calls = client.head_calls + # Overwritten object -> new backend ETag -> must re-resolve, not reuse. + asyncio.run(handler._process_list_objects(client, "bucket", [_obj("mp.db", 88888, "v2")])) + + assert client.head_calls == head_calls + 1 + + +def test_failed_head_falls_back_and_does_not_cache(fresh_cache): + handler = FakeHandler() + client = FakeClient(fail_head_key="broken.db") + + result = asyncio.run( + handler._process_list_objects(client, "bucket", [_obj("broken.db", size=555)]) + ) + + assert result[0]["size"] == 555 + assert result[0]["etag"] == "backend-etag" + assert len(fresh_cache) == 0 + + +def test_cache_evicts_least_recently_used(): + cache = PlaintextAttrCache(maxsize=2) + cache.put("b", "k1", "e1", 1, "t1") + cache.put("b", "k2", "e2", 2, "t2") + assert cache.get("b", "k1", "e1") == (1, "t1") # refresh k1 + cache.put("b", "k3", "e3", 3, "t3") # evicts k2 + + assert cache.get("b", "k2", "e2") is None + assert cache.get("b", "k1", "e1") == (1, "t1") + assert cache.get("b", "k3", "e3") == (3, "t3") + + +def test_cache_ignores_empty_backend_etag(): + cache = PlaintextAttrCache() + cache.put("b", "k", "", 1, "t") + assert cache.get("b", "k", "") is None + assert len(cache) == 0 + + +class _FakeURL: + def __init__(self, path, query): + self.path = path + self.query = query + + +class _FakeRequest: + def __init__(self, path, query, body): + self.url = _FakeURL(path, query) + self.headers = {} + self._body = body + + async def body(self): + return self._body + + +class _ClientCM: + def __init__(self, client): + self._client = client + + async def __aenter__(self): + return self._client + + async def __aexit__(self, *exc): + return False + + +async def test_complete_multipart_primes_cache( + handler, mock_s3, mock_s3_client, settings, credentials +): + bucket, key = "test-bucket", "backups/primed.bin" + kid, kek = settings.keyring.key_for(credentials.access_key) + dek = crypto.generate_dek() + plaintext = b"payload" * 128 + + resp = await mock_s3.create_multipart_upload(bucket, key) + upload_id = resp["UploadId"] + ciphertext = crypto.encrypt(plaintext, dek, crypto.derive_part_nonce(upload_id, 1)) + await mock_s3.upload_part(bucket, key, upload_id, 1, ciphertext) + await persist_upload_state( + mock_s3_client, bucket, key, upload_id, crypto.wrap_key(dek, kek), kid + ) + handler._client = lambda creds: _ClientCM(mock_s3_client) + + body = ( + "" + "1"etag-1"" + "" + ).encode() + resp = await handler.handle_complete_multipart_upload( + _FakeRequest(f"/{bucket}/{key}", urlencode({"uploadId": upload_id}), body), + credentials, + ) + assert resp.status_code == 200 + + backend_etag = str(mock_s3.objects[f"{bucket}/{key}"]["ETag"]).strip('"') + assert plaintext_attr_cache.get(bucket, key, backend_etag) == ( + len(plaintext), + synthetic_multipart_etag(len(plaintext)), + ) diff --git a/tests/unit/test_list_objects_parallel.py b/tests/unit/test_list_objects_parallel.py index f8cea4a..2160633 100644 --- a/tests/unit/test_list_objects_parallel.py +++ b/tests/unit/test_list_objects_parallel.py @@ -15,6 +15,7 @@ class FakeHandler: _process_list_objects = BucketHandlerMixin._process_list_objects + _list_entry = staticmethod(BucketHandlerMixin._list_entry) def __init__(self): self.inflight = 0 From ebd6f6571131e4cb267770725acda22bbfb61948 Mon Sep 17 00:00:00 2001 From: serversidehannes Date: Sun, 12 Jul 2026 18:07:45 +0200 Subject: [PATCH 2/2] style: use bytes literal for complete request body in test --- tests/unit/test_list_multipart_plaintext_size.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/tests/unit/test_list_multipart_plaintext_size.py b/tests/unit/test_list_multipart_plaintext_size.py index 19fe598..981fa58 100644 --- a/tests/unit/test_list_multipart_plaintext_size.py +++ b/tests/unit/test_list_multipart_plaintext_size.py @@ -254,10 +254,10 @@ async def test_complete_multipart_primes_cache( handler._client = lambda creds: _ClientCM(mock_s3_client) body = ( - "" - "1"etag-1"" - "" - ).encode() + b"" + b"1"etag-1"" + b"" + ) resp = await handler.handle_complete_multipart_upload( _FakeRequest(f"/{bucket}/{key}", urlencode({"uploadId": upload_id}), body), credentials,