diff --git a/tests/test_storage_request_retry.py b/tests/test_storage_request_retry.py new file mode 100644 index 00000000..ec6e9476 --- /dev/null +++ b/tests/test_storage_request_retry.py @@ -0,0 +1,175 @@ +# Copyright 2025 Huawei Technologies Co., Ltd. All Rights Reserved. +# Copyright 2025 The TransferQueue Team +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Storage-unit retry and timeout-diagnosis tests.""" + +import logging +from unittest.mock import AsyncMock, MagicMock, patch + +import numpy as np +import pytest +import torch +import zmq +from tensordict import TensorDict + +from transfer_queue.metadata import BatchMeta +from transfer_queue.storage.managers import simple_storage_manager as ssm +from transfer_queue.storage.managers.simple_storage_manager import AsyncSimpleStorageManager, StorageUnitTimeout +from transfer_queue.utils import common +from transfer_queue.utils.enum_utils import Role +from transfer_queue.utils.zmq_utils import ZMQServerInfo + + +def _manager(with_unit: bool = True) -> AsyncSimpleStorageManager: + manager = AsyncSimpleStorageManager.__new__(AsyncSimpleStorageManager) + manager.storage_manager_id = "TQ_STORAGE_test" + manager.storage_unit_infos = {} + if with_unit: + manager.storage_unit_infos["unit_a"] = ZMQServerInfo( + role=Role.STORAGE, + id="unit_a", + ip="10.0.0.7", + ports={"put_get_socket": 5555}, + ) + manager.close = lambda: None + return manager + + +def _single_sample_batch() -> tuple[TensorDict, BatchMeta]: + metadata = BatchMeta( + global_indexes=[0], + partition_ids=["0"], + field_schema={"input_ids": {"dtype": torch.int64, "shape": (2,), "is_nested": False, "is_non_tensor": False}}, + production_status=np.ones(1, dtype=np.int8), + ) + return TensorDict({"input_ids": torch.zeros(1, 2, dtype=torch.int64)}, batch_size=1), metadata + + +@pytest.mark.asyncio +@pytest.mark.parametrize("data_parser, expected_attempts", [(None, 3), (lambda data: data, 1)]) +async def test_put_retries_only_without_a_parser(data_parser, expected_attempts): + manager = _manager() + manager.notify_data_update = AsyncMock() + manager._put_to_single_storage_unit = AsyncMock(side_effect=StorageUnitTimeout("no answer")) + data, metadata = _single_sample_batch() + + with ( + patch.object(ssm, "TQ_SIMPLE_STORAGE_MAX_ATTEMPTS", 3), + patch.object(manager, "_diagnose_storage_unit", return_value="diagnosis"), + pytest.raises(StorageUnitTimeout), + ): + await manager.put_data(data, metadata, data_parser=data_parser) + + assert manager._put_to_single_storage_unit.await_count == expected_attempts + manager.notify_data_update.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_lost_request_recovers_without_diagnosis(): + manager = _manager() + request = AsyncMock(side_effect=[StorageUnitTimeout("no answer"), "payload"]) + + with patch.object(manager, "_diagnose_storage_unit") as diagnose: + result = await manager._request_with_retry("get", "unit_a", "samples=4", request) + + assert result == "payload" + assert request.await_count == 2 + diagnose.assert_not_awaited() + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "error, max_attempts, expected_attempts", + [ + (StorageUnitTimeout("no answer"), 3, 3), + (RuntimeError("unit rejected request"), 3, 1), + (StorageUnitTimeout("no answer"), 0, 1), + ], +) +async def test_retry_attempts_are_bounded_and_only_cover_timeouts(error, max_attempts, expected_attempts): + manager = _manager() + request = AsyncMock(side_effect=error) + + with ( + patch.object(manager, "_diagnose_storage_unit", return_value="diagnosis"), + pytest.raises(type(error)), + ): + await manager._request_with_retry("get", "unit_a", "samples=4", request, max_attempts=max_attempts) + + assert request.await_count == expected_attempts + + +@pytest.mark.asyncio +async def test_failure_log_has_context_without_repeating_the_unit(caplog): + manager = _manager() + error = StorageUnitTimeout("no answer in 200s during get from storage unit unit_a at 10.0.0.7:5555") + + with ( + patch.object(manager, "_diagnose_storage_unit", return_value="diagnosis"), + caplog.at_level(logging.ERROR), + pytest.raises(StorageUnitTimeout), + ): + await manager._request_with_retry( + "get", "unit_a", "samples=4 fields=['input_ids']", AsyncMock(side_effect=error) + ) + + message = next(record.message for record in caplog.records if "failed after" in record.message) + assert "samples=4 fields=['input_ids']" in message + assert message.count("unit_a") == message.count("10.0.0.7:5555") == 1 + + +@pytest.mark.parametrize( + "elapsed, payload_bytes, should_log", + [(0.01, 2**20, False), (0.01, 512 * 2**20, True), (6.0, 2**20, True)], +) +def test_log_heavy_operation_thresholds(caplog, elapsed, payload_bytes, should_log): + with caplog.at_level(logging.WARNING): + common.log_heavy_operation("TQ_STORAGE_test", "put", elapsed, payload_bytes, "samples=4") + + assert bool(caplog.records) is should_log + if should_log: + assert "serialized_mb=" in caplog.records[0].message + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "tcp_result, probe_result, expected", + [ + ((None, None), {"active_keys": 4, "op_stats": {"GET_DATA": {"request_count": 1025}}}, "request_lost"), + ((None, None), zmq.error.Again(), "unit_not_serving"), + (ConnectionRefusedError(), zmq.error.Again(), "tcp=down(ConnectionRefusedError)"), + ], +) +async def test_diagnosis_classifies_the_failure(tcp_result, probe_result, expected): + manager = _manager() + writer = MagicMock() + tcp = AsyncMock( + side_effect=tcp_result if isinstance(tcp_result, Exception) else None, + return_value=(None, writer), + ) + probe = AsyncMock( + side_effect=probe_result if isinstance(probe_result, Exception) else None, + return_value=probe_result, + ) + + with patch.object(ssm.asyncio, "open_connection", tcp), patch.object(manager, "_probe_storage_unit", probe): + diagnosis = await manager._diagnose_storage_unit("unit_a") + + assert expected in diagnosis + + +@pytest.mark.asyncio +async def test_diagnosis_handles_an_unknown_unit(): + assert "unit_not_registered" in await _manager(with_unit=False)._diagnose_storage_unit("missing") diff --git a/transfer_queue/storage/managers/simple_storage_manager.py b/transfer_queue/storage/managers/simple_storage_manager.py index b986701a..70765cb9 100644 --- a/transfer_queue/storage/managers/simple_storage_manager.py +++ b/transfer_queue/storage/managers/simple_storage_manager.py @@ -15,9 +15,11 @@ import asyncio import os +import time import warnings from collections import defaultdict from collections.abc import Mapping +from functools import partial from operator import itemgetter from pathlib import Path from typing import Any, Callable, NamedTuple @@ -29,7 +31,11 @@ from transfer_queue.metadata import BatchMeta, extract_field_schema from transfer_queue.storage.managers.base import StorageManager, StorageManagerFactory -from transfer_queue.storage.simple_storage import KEY_NOT_FOUND_MARKER, StorageKeyNotFoundError +from transfer_queue.storage.simple_storage import ( + KEY_NOT_FOUND_MARKER, + StorageKeyNotFoundError, +) +from transfer_queue.utils.common import log_heavy_operation from transfer_queue.utils.logging_utils import get_logger from transfer_queue.utils.zmq_utils import ( TQ_SOCKET_POOL_SIZE, @@ -37,6 +43,7 @@ ZMQRequestType, ZMQServerInfo, ZMQSocketPool, + frame_nbytes, with_zmq_socket, ) @@ -44,6 +51,23 @@ TQ_SIMPLE_STORAGE_SEND_RECV_TIMEOUT = int(os.environ.get("TQ_SIMPLE_STORAGE_SEND_RECV_TIMEOUT", 200)) # seconds +# Attempts per storage-unit request, including the first. Each one must use a new connection: a +# DEALER silently queues messages for a peer it never reached, so no timeout can catch that. +TQ_SIMPLE_STORAGE_MAX_ATTEMPTS = int(os.environ.get("TQ_SIMPLE_STORAGE_MAX_ATTEMPTS", 3)) + +# Timeout for the post-failure probe, which only has to answer whether the unit still serves. +TQ_SIMPLE_STORAGE_PROBE_TIMEOUT = int(os.environ.get("TQ_SIMPLE_STORAGE_PROBE_TIMEOUT", 10)) + + +class StorageUnitTimeout(RuntimeError): + """A storage unit did not answer within the send/recv timeout. + + Distinct from an error the unit reported: only a missing answer is worth a new connection. + The message must name the unit, its endpoint and the timeout, because it is what the callers + of ``put_data`` and ``get_data`` see, and the retry logs rely on it instead of repeating them. + """ + + _SU_SUBDIR = "simple_storage" _SU_INFO_FILE = "storage_unit_info.json" @@ -54,6 +78,13 @@ resolve_target=lambda args, kwargs: kwargs.get("target_storage_unit"), ) +# Same endpoint as above but with the short diagnostic timeout, used only after a failure. +with_storage_unit_probe_socket = with_zmq_socket( + get_peer=lambda self, target: self.storage_unit_infos[target], + get_pool=lambda self: self.storage_probe_pool, + resolve_target=lambda args, kwargs: kwargs.get("target_storage_unit"), +) + class RoutingGroup(NamedTuple): """Routing result for a single storage unit.""" @@ -84,6 +115,13 @@ def __init__( "put_get_socket", timeout=TQ_SIMPLE_STORAGE_SEND_RECV_TIMEOUT, ) + self.storage_probe_pool = ZMQSocketPool( + self.zmq_context, + f"{self.storage_manager_id}_probe", + "put_get_socket", + timeout=TQ_SIMPLE_STORAGE_PROBE_TIMEOUT, + maxsize=1, + ) self.config = config server_infos: ZMQServerInfo | dict[str, ZMQServerInfo] | None = config.get("zmq_info", None) @@ -114,11 +152,11 @@ def _warn_if_pool_can_exhaust_context(self, num_units: int) -> None: budget = self.zmq_context.get(zmq.MAX_SOCKETS) except zmq.ZMQError: # pragma: no cover - context already terminating return - worst_case = TQ_SOCKET_POOL_SIZE * num_units + worst_case = (TQ_SOCKET_POOL_SIZE + 1) * num_units if worst_case > budget: logger.warning( f"[{self.storage_manager_id}]: storage RPC pool may hold up to " - f"{TQ_SOCKET_POOL_SIZE} x {num_units} = {worst_case} idle sockets, above this " + f"({TQ_SOCKET_POOL_SIZE} + 1 probe) x {num_units} = {worst_case} idle sockets, above this " f"context's ZMQ_MAX_SOCKETS ({budget}). Concurrent requests can then fail to " f"open a socket. Lower TQ_SOCKET_POOL_SIZE or raise TQ_CLIENT_ZMQ_MAX_SOCKETS " f"(with enough file descriptors, see ulimit -n)." @@ -170,6 +208,108 @@ def _group_by_hash(self, global_indexes: list[int]) -> dict[str, RoutingGroup]: pos_lists[key].append(pos) return {key: RoutingGroup(gi_lists[key], pos_lists[key]) for key in gi_lists} + def _describe_storage_unit(self, storage_unit_id: str) -> str: + """Return ``ip:port`` for a storage unit, for use in diagnostics. + + The unit id is a random uuid4 fragment that carries no location, so a bare id in an + error message cannot be traced back to a node. Never raises: it is only ever called + while reporting another failure, and must not mask it. + """ + info = self.storage_unit_infos.get(storage_unit_id) + if info is None: + return "endpoint unknown (unit not registered with this manager)" + return f"{info.ip}:{info.ports.get('put_get_socket')}" + + @with_storage_unit_probe_socket + async def _probe_storage_unit(self, target_storage_unit: str, socket: zmq.Socket = None) -> dict[str, Any]: + """Ask a storage unit for its own counters over a brand-new socket. + + Served by the same worker thread as put and get, so an answer proves the unit is serving. + """ + request_msg = ZMQMessage.create( + request_type=ZMQRequestType.GET_METRICS, # type: ignore[arg-type] + sender_id=f"{self.storage_manager_id}_probe", + receiver_id=target_storage_unit, + body={}, + ) + await socket.send_multipart(request_msg.serialize()) + messages = await socket.recv_multipart(copy=False) + response_msg = ZMQMessage.deserialize(messages) + if response_msg.request_type != ZMQRequestType.METRICS_RESPONSE: + raise RuntimeError(f"unexpected probe response type {response_msg.request_type}") + return response_msg.body + + async def _diagnose_storage_unit(self, target_storage_unit: str) -> str: + """Classify a timeout as a lost request, a stuck unit, or an unreachable node. + + Returns one log line and never raises: it runs while another failure is being reported. + """ + info = self.storage_unit_infos.get(target_storage_unit) + if info is None: + return "verdict=unknown(unit_not_registered)" + + try: + _, writer = await asyncio.wait_for( + asyncio.open_connection(info.ip, info.ports.get("put_get_socket")), timeout=5 + ) + writer.close() + tcp = "tcp=up" + except Exception as e: + tcp = f"tcp=down({type(e).__name__})" + + try: + body = await self._probe_storage_unit(target_storage_unit=target_storage_unit) + op_counts = {op: stats.get("request_count") for op, stats in (body.get("op_stats") or {}).items()} + return ( + f"{tcp} verdict=request_lost_in_flight (unit answered a fresh probe: " + f"ops={op_counts} active_keys={body.get('active_keys')} " + f"rss_gb={body.get('process_rss_bytes', 0) / 2**30:.2f})" + ) + except zmq.error.Again: + return f"{tcp} verdict=unit_not_serving (no probe answer in {TQ_SIMPLE_STORAGE_PROBE_TIMEOUT}s)" + except Exception as e: + return f"{tcp} verdict=unknown (probe failed: {type(e).__name__}: {e})" + + async def _request_with_retry( + self, + operation: str, + target_storage_unit: str, + request_context: str, + make_request: Callable[[], Any], + max_attempts: int | None = None, + ): + """Run one storage-unit request, retrying a missing answer on a fresh connection. + + Args: + operation: Operation name for logs, e.g. ``get`` or ``put``. + target_storage_unit: Unit this request is routed to. + request_context: Request shape, so both ends of a failure can be correlated. + make_request: Zero-arg callable returning a coroutine for one attempt. Must build a + new socket per call, which ``with_storage_unit_socket`` does. + max_attempts: Attempts allowed. Defaults to ``TQ_SIMPLE_STORAGE_MAX_ATTEMPTS``; pass 1 + for a request that must not be replayed. + """ + # Floored at one: a nonpositive count would skip the request and report success, which for + # put would publish metadata for data that was never sent. + attempts_allowed = max(1, TQ_SIMPLE_STORAGE_MAX_ATTEMPTS if max_attempts is None else max_attempts) + for attempt in range(1, attempts_allowed + 1): + try: + return await make_request() + except StorageUnitTimeout as e: + # The exception already names the unit, its endpoint and the timeout, so these + # lines only add what it cannot know: which attempt this was, and the shape. + if attempt < attempts_allowed: + logger.warning( + f"[{self.storage_manager_id}]: {operation} retry {attempt + 1}/{attempts_allowed} " + f"on a new connection. {request_context} {e}" + ) + continue + logger.error( + f"[{self.storage_manager_id}]: {operation} failed after {attempt} attempts. " + f"{request_context} {e} {await self._diagnose_storage_unit(target_storage_unit)}" + ) + raise + @staticmethod def _select_by_positions(field_data, positions: list[int]): """Slice a single field's data by non-contiguous batch positions. @@ -283,24 +423,37 @@ async def put_data( field_schema = extract_field_schema(data) routing = self._group_by_hash(metadata.global_indexes) - tasks = [ - self._put_to_single_storage_unit( - group.global_indexes, - {f: self._select_by_positions(data[f], group.batch_positions) for f in data.keys()}, - target_storage_unit=su_id, - data_parser=data_parser, + # Parser-backed puts are not replayed: the public API does not constrain parser side effects. + max_attempts = 1 if data_parser is not None else None + tasks = [] + for su_id, group in routing.items(): + storage_data = {f: self._select_by_positions(data[f], group.batch_positions) for f in data.keys()} + tasks.append( + self._request_with_retry( + "put", + su_id, + f"samples={len(group.global_indexes)} fields={list(storage_data.keys())}", + partial( + self._put_to_single_storage_unit, + group.global_indexes, + storage_data, + target_storage_unit=su_id, + data_parser=data_parser, + ), + max_attempts=max_attempts, + ) ) - for su_id, group in routing.items() - ] try: await asyncio.gather(*tasks) except Exception as e: + # The offending unit is named in the error itself; the full routing list can run to + # hundreds of ids, which buries it. logger.error( f"[{self.storage_manager_id}]: put_data failed. " f"partition_id={metadata.partition_ids[0]}, " f"num_samples={metadata.size}, " - f"storage_units={list(routing.keys())}, " + f"num_storage_units={len(routing)}, " f"error={type(e).__name__}: {e}" ) raise @@ -332,9 +485,12 @@ async def _put_to_single_storage_unit( body={"global_indexes": global_indexes, "data": storage_data, "data_parser": data_parser}, ) + serialized_bytes = 0 + started = time.perf_counter() try: - data = request_msg.serialize() - await socket.send_multipart(data, copy=False) + frames = request_msg.serialize() + serialized_bytes = sum(frame_nbytes(frame) or 0 for frame in frames) + await socket.send_multipart(frames, copy=False) messages = await socket.recv_multipart(copy=False) response_msg = ZMQMessage.deserialize(messages) @@ -343,15 +499,21 @@ async def _put_to_single_storage_unit( f"Failed to put data to storage unit {target_storage_unit}: " f"{response_msg.body.get('message', 'Unknown error')}" ) - except zmq.error.Again as e: - timeout_sec = TQ_SIMPLE_STORAGE_SEND_RECV_TIMEOUT - logger.error( - f"[{self.storage_manager_id}]: ZMQ recv timeout ({timeout_sec}s) " - f"during put to storage unit {target_storage_unit}. " - f"The storage unit may be overloaded or crashed." + # This end serializes the put payload and waits out the round trip, so it holds both + # the true wire size and the end-to-end latency. + log_heavy_operation( + self.storage_manager_id, + "put", + time.perf_counter() - started, + serialized_bytes, + f"to {target_storage_unit} at {self._describe_storage_unit(target_storage_unit)} " + f"samples={len(global_indexes)} fields={list(storage_data.keys())}", ) - raise RuntimeError( - f"ZMQ recv timeout ({timeout_sec}s) during put to storage unit {target_storage_unit}" + except zmq.error.Again as e: + raise StorageUnitTimeout( + f"no answer in {TQ_SIMPLE_STORAGE_SEND_RECV_TIMEOUT}s during put to storage unit " + f"{target_storage_unit} at {self._describe_storage_unit(target_storage_unit)}; " + f"serialized_mb={serialized_bytes / 2**20:.1f}" ) from e except Exception as e: logger.error( @@ -431,7 +593,17 @@ async def get_data(self, metadata: BatchMeta) -> TensorDict: routing = self._group_by_hash(metadata.global_indexes) tasks = [ - self._get_from_single_storage_unit(group.global_indexes, metadata.field_names, target_storage_unit=su_id) + self._request_with_retry( + "get", + su_id, + f"samples={len(group.global_indexes)} fields={list(metadata.field_names)}", + partial( + self._get_from_single_storage_unit, + group.global_indexes, + metadata.field_names, + target_storage_unit=su_id, + ), + ) for su_id, group in routing.items() ] try: @@ -490,13 +662,10 @@ async def _get_from_single_storage_unit( error_type = StorageKeyNotFoundError if KEY_NOT_FOUND_MARKER in message else RuntimeError raise error_type(f"Failed to get data from storage unit {target_storage_unit}: {message}") except zmq.error.Again as e: - timeout_sec = TQ_SIMPLE_STORAGE_SEND_RECV_TIMEOUT - logger.error( - f"[{self.storage_manager_id}]: ZMQ recv timeout ({timeout_sec}s) " - f"from storage unit {target_storage_unit}. " - f"The storage unit may be overloaded or crashed." - ) - raise RuntimeError(f"ZMQ recv timeout ({timeout_sec}s) from storage unit {target_storage_unit}") from e + raise StorageUnitTimeout( + f"no answer in {TQ_SIMPLE_STORAGE_SEND_RECV_TIMEOUT}s during get from storage unit " + f"{target_storage_unit} at {self._describe_storage_unit(target_storage_unit)}" + ) from e except StorageKeyNotFoundError: # Already logged at debug by the storage unit; propagate for the caller to classify. raise @@ -690,4 +859,7 @@ def close(self) -> None: pool = getattr(self, "storage_rpc_pool", None) if pool is not None: pool.close() + probe_pool = getattr(self, "storage_probe_pool", None) + if probe_pool is not None: + probe_pool.close() super().close() diff --git a/transfer_queue/storage/simple_storage.py b/transfer_queue/storage/simple_storage.py index bf9d616f..cd0d2b4b 100644 --- a/transfer_queue/storage/simple_storage.py +++ b/transfer_queue/storage/simple_storage.py @@ -25,7 +25,7 @@ import ray import zmq -from transfer_queue.utils.common import limit_pytorch_auto_parallel_threads +from transfer_queue.utils.common import limit_pytorch_auto_parallel_threads, log_heavy_operation from transfer_queue.utils.enum_utils import Role from transfer_queue.utils.logging_utils import get_logger from transfer_queue.utils.perf_utils import IntervalPerfMonitor @@ -36,6 +36,7 @@ ZMQServerInfo, create_zmq_socket, format_zmq_address, + frame_nbytes, get_free_port, get_node_ip_address, ) @@ -344,6 +345,7 @@ def _worker_routine(self) -> None: worker_socket.send_multipart([identity] + error_msg.serialize(), copy=False) continue operation = request_msg.request_type + started = time.perf_counter() try: logger.debug(f"[{self.storage_unit_id}]: worker received operation: {operation}") @@ -388,7 +390,18 @@ def _worker_routine(self) -> None: ) # Send response back with identity for routing - worker_socket.send_multipart([identity] + response_msg.serialize(), copy=False) + response_frames = response_msg.serialize() + if operation == ZMQRequestType.GET_DATA: # type: ignore[arg-type] + # This end serializes the get response, so its frames give the true wire size. + log_heavy_operation( + self.storage_unit_id, + "get", + time.perf_counter() - started, + sum(frame_nbytes(frame) or 0 for frame in response_frames), + f"samples={len(request_msg.body.get('global_indexes', []))} " + f"fields={list(request_msg.body.get('fields', []))}", + ) + worker_socket.send_multipart([identity] + response_frames, copy=False) logger.info(f"[{self.storage_unit_id}]: worker stopped.") poller.unregister(worker_socket) diff --git a/transfer_queue/utils/common.py b/transfer_queue/utils/common.py index 999e8183..1278dfd4 100644 --- a/transfer_queue/utils/common.py +++ b/transfer_queue/utils/common.py @@ -133,3 +133,27 @@ def get_env_bool(env_key: str, default: bool = False) -> bool: true_values = {"true", "1", "yes", "y", "on"} return env_value_lower in true_values + + +# Thresholds above which a single request earns a log line. Both sit far above the normal range +# of single-digit milliseconds, so tripping either one is itself the finding. +TQ_STORAGE_SLOW_REQUEST_SECONDS = float(os.environ.get("TQ_STORAGE_SLOW_REQUEST_SECONDS", 5.0)) +TQ_STORAGE_LARGE_PAYLOAD_MB = float(os.environ.get("TQ_STORAGE_LARGE_PAYLOAD_MB", 256)) + + +def log_heavy_operation(component_id: str, operation: str, elapsed: float, payload_bytes: int, detail: str) -> None: + """Warn about one slow or unusually large request; stay silent otherwise. + + Args: + component_id: Storage manager or storage unit reporting the request. + operation: Operation name, e.g. ``put`` or ``get``. + elapsed: Wall time spent on the request, in seconds. + payload_bytes: Serialized size of the payload on the wire. + detail: Request shape, appended to the message verbatim. + """ + payload_mb = payload_bytes / 2**20 + if elapsed < TQ_STORAGE_SLOW_REQUEST_SECONDS and payload_mb < TQ_STORAGE_LARGE_PAYLOAD_MB: + return + logger.warning( + f"[{component_id}]: heavy {operation} {detail} serialized_mb={payload_mb:.1f} elapsed={elapsed:.2f}s" + )