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..981fa58 --- /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 = ( + b"" + b"1"etag-1"" + b"" + ) + 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