⚡️ Speed up method RetryOnRpcErrorClientInterceptor._is_retryable_error by 116%

[codeflash-ai]

📄 116% (1.16x) speedup for RetryOnRpcErrorClientInterceptor._is_retryable_error in pinecone/grpc/retry.py

⏱️ Runtime : 2.23 milliseconds → 1.03 milliseconds (best of 242 runs)

📝 Explanation and details

The optimization restructures the conditional logic to eliminate expensive operations on the critical path. The key improvements are:

1. Early Return Pattern: The optimized version uses explicit early returns instead of a compound boolean expression, which reduces unnecessary evaluations when isinstance() fails.

2. Efficient Class Name Access: Replaces the expensive response_or_error.__class__.__name__ with type(response_or_error).__name__, which is faster for attribute access.

3. Reduced String Operations: By separating the conditions, the "_MultiThreadedRendezvous" in cls_name check only executes for actual RpcError instances, avoiding the string operation for non-RpcError objects entirely.

4. Optimized Control Flow: The sequential if-statements create better branch prediction opportunities and allow the CPU to skip expensive operations when early conditions fail.

The line profiler shows the most dramatic improvement for _MultiThreadedRendezvous errors (699% faster) and custom class name scenarios (500%+ faster), where the original version performed redundant __class__.__name__ lookups. For common cases like basic retryable errors, the speedup is more modest (6-24%) but consistent.

This optimization is particularly effective for workloads with many non-RpcError objects or _MultiThreadedRendezvous instances, where the early exit paths provide substantial performance gains.

✅ Correctness verification report:

Test	Status
⚙️ Existing Unit Tests	🔘 None Found
🌀 Generated Regression Tests	✅ 3173 Passed
⏪ Replay Tests	🔘 None Found
🔎 Concolic Coverage Tests	🔘 None Found
📊 Tests Coverage	100.0%

🌀 Generated Regression Tests and Runtime

import grpc
# imports
import pytest
from pinecone.grpc.retry import RetryOnRpcErrorClientInterceptor


# Helper classes for testing
class DummyRetryConfig:
    def __init__(self, max_attempts=3, sleep_policy=None, retryable_status=None):
        self.max_attempts = max_attempts
        self.sleep_policy = sleep_policy
        self.retryable_status = retryable_status or set()

class DummyRpcError(grpc.RpcError):
    """A dummy RpcError for testing with a customizable code and class name."""
    def __init__(self, code):
        self._code = code

    def code(self):
        return self._code

class DummyNonRetryableRpcError(grpc.RpcError):
    """A dummy RpcError for testing with a code not in retryable_status."""
    def __init__(self, code):
        self._code = code

    def code(self):
        return self._code

class DummyMultiThreadedRendezvous(grpc.RpcError):
    """A dummy class mimicking _MultiThreadedRendezvous."""
    def __init__(self, code):
        self._code = code

    def code(self):
        return self._code

    @property
    def __class__(self):
        class _Fake:
            __name__ = "_MultiThreadedRendezvous"
        return _Fake

# -------------------- UNIT TESTS --------------------

# ---- BASIC TEST CASES ----

def test_retryable_error_basic():
    """Test a basic retryable error (RpcError, not _MultiThreadedRendezvous, code in retryable_status)."""
    retryable_codes = {grpc.StatusCode.UNAVAILABLE, grpc.StatusCode.DEADLINE_EXCEEDED}
    interceptor = RetryOnRpcErrorClientInterceptor(
        DummyRetryConfig(retryable_status=retryable_codes)
    )
    err = DummyRpcError(grpc.StatusCode.UNAVAILABLE)
    codeflash_output = interceptor._is_retryable_error(err) # 1.88μs -> 1.52μs (23.8% faster)

def test_non_retryable_error_code_not_in_retryable_status():
    """Test RpcError where code is NOT in retryable_status."""
    retryable_codes = {grpc.StatusCode.UNAVAILABLE}
    interceptor = RetryOnRpcErrorClientInterceptor(
        DummyRetryConfig(retryable_status=retryable_codes)
    )
    err = DummyNonRetryableRpcError(grpc.StatusCode.PERMISSION_DENIED)
    codeflash_output = interceptor._is_retryable_error(err) # 1.74μs -> 1.60μs (8.43% faster)

def test_non_rpc_error_type():
    """Test input that is NOT an instance of grpc.RpcError."""
    retryable_codes = {grpc.StatusCode.UNAVAILABLE}
    interceptor = RetryOnRpcErrorClientInterceptor(
        DummyRetryConfig(retryable_status=retryable_codes)
    )
    not_error = Exception("Not an RpcError")
    codeflash_output = interceptor._is_retryable_error(not_error) # 737ns -> 732ns (0.683% faster)

def test_multithreaded_rendezvous_error():
    """Test that _MultiThreadedRendezvous errors are NOT retryable, even if code matches."""
    retryable_codes = {grpc.StatusCode.UNAVAILABLE}
    interceptor = RetryOnRpcErrorClientInterceptor(
        DummyRetryConfig(retryable_status=retryable_codes)
    )
    err = DummyMultiThreadedRendezvous(grpc.StatusCode.UNAVAILABLE)
    codeflash_output = interceptor._is_retryable_error(err) # 12.8μs -> 1.60μs (699% faster)

# ---- EDGE TEST CASES ----

def test_retryable_status_empty():
    """Test with empty retryable_status set: no errors should be retryable."""
    interceptor = RetryOnRpcErrorClientInterceptor(
        DummyRetryConfig(retryable_status=set())
    )
    err = DummyRpcError(grpc.StatusCode.UNAVAILABLE)
    codeflash_output = interceptor._is_retryable_error(err) # 1.71μs -> 1.60μs (6.69% faster)

def test_code_method_raises():
    """Test error object whose code() method raises an exception."""
    class BadRpcError(grpc.RpcError):
        def code(self):
            raise RuntimeError("code method failed")
    retryable_codes = {grpc.StatusCode.UNAVAILABLE}
    interceptor = RetryOnRpcErrorClientInterceptor(
        DummyRetryConfig(retryable_status=retryable_codes)
    )
    err = BadRpcError()
    # Should raise, not return True/False
    with pytest.raises(RuntimeError):
        interceptor._is_retryable_error(err) # 1.97μs -> 1.72μs (14.9% faster)

def test_error_with_class_name_similar_to_multithreaded_rendezvous():
    """Test error whose class name contains '_MultiThreadedRendezvous' as a substring."""
    class SimilarNameError(grpc.RpcError):
        def code(self):
            return grpc.StatusCode.UNAVAILABLE
        @property
        def __class__(self):
            class _Fake:
                __name__ = "Custom_MultiThreadedRendezvousError"
            return _Fake
    retryable_codes = {grpc.StatusCode.UNAVAILABLE}
    interceptor = RetryOnRpcErrorClientInterceptor(
        DummyRetryConfig(retryable_status=retryable_codes)
    )
    err = SimilarNameError()
    # Should NOT be retryable due to substring match
    codeflash_output = interceptor._is_retryable_error(err) # 10.6μs -> 1.70μs (524% faster)

def test_error_with_class_name_not_multithreaded_rendezvous():
    """Test error whose class name is similar but does NOT contain the forbidden substring."""
    class NotRendezvous(grpc.RpcError):
        def code(self):
            return grpc.StatusCode.UNAVAILABLE
        @property
        def __class__(self):
            class _Fake:
                __name__ = "MultiThreadedRendezvousCustom"
            return _Fake
    retryable_codes = {grpc.StatusCode.UNAVAILABLE}
    interceptor = RetryOnRpcErrorClientInterceptor(
        DummyRetryConfig(retryable_status=retryable_codes)
    )
    err = NotRendezvous()
    # Should be retryable because substring is not exact
    codeflash_output = interceptor._is_retryable_error(err) # 9.47μs -> 1.58μs (500% faster)

def test_error_with_none_code():
    """Test error whose code() returns None."""
    class NoneCodeError(grpc.RpcError):
        def code(self):
            return None
    retryable_codes = {grpc.StatusCode.UNAVAILABLE}
    interceptor = RetryOnRpcErrorClientInterceptor(
        DummyRetryConfig(retryable_status=retryable_codes)
    )
    err = NoneCodeError()
    codeflash_output = interceptor._is_retryable_error(err) # 1.38μs -> 1.18μs (17.4% faster)

def test_error_with_non_status_code():
    """Test error whose code() returns a non-StatusCode value."""
    class WeirdCodeError(grpc.RpcError):
        def code(self):
            return "UNAVAILABLE"
    retryable_codes = {grpc.StatusCode.UNAVAILABLE}
    interceptor = RetryOnRpcErrorClientInterceptor(
        DummyRetryConfig(retryable_status=retryable_codes)
    )
    err = WeirdCodeError()
    codeflash_output = interceptor._is_retryable_error(err) # 1.45μs -> 1.24μs (16.7% faster)

# ---- LARGE SCALE TEST CASES ----

def test_many_retryable_status_codes():
    """Test with a large set of retryable_status codes and many error instances."""
    all_codes = set(grpc.StatusCode)
    interceptor = RetryOnRpcErrorClientInterceptor(
        DummyRetryConfig(retryable_status=all_codes)
    )
    # Test all codes: should be retryable for all
    for code in all_codes:
        err = DummyRpcError(code)
        codeflash_output = interceptor._is_retryable_error(err) # 8.37μs -> 8.27μs (1.19% faster)

def test_many_errors_mixed_types():
    """Test a large number of mixed error objects, some retryable, some not."""
    retryable_codes = {grpc.StatusCode.UNAVAILABLE, grpc.StatusCode.DEADLINE_EXCEEDED}
    interceptor = RetryOnRpcErrorClientInterceptor(
        DummyRetryConfig(retryable_status=retryable_codes)
    )
    # Create 500 retryable and 500 non-retryable errors
    errors = [
        DummyRpcError(grpc.StatusCode.UNAVAILABLE) for _ in range(250)
    ] + [
        DummyRpcError(grpc.StatusCode.DEADLINE_EXCEEDED) for _ in range(250)
    ] + [
        DummyRpcError(grpc.StatusCode.PERMISSION_DENIED) for _ in range(250)
    ] + [
        DummyMultiThreadedRendezvous(grpc.StatusCode.UNAVAILABLE) for _ in range(250)
    ]
    # First 500 should be retryable, last 500 should not
    for i, err in enumerate(errors):
        if i < 500:
            codeflash_output = interceptor._is_retryable_error(err)
        else:
            codeflash_output = interceptor._is_retryable_error(err)

def test_large_non_rpc_error_list():
    """Test a large number of non-RpcError objects to ensure always False."""
    retryable_codes = {grpc.StatusCode.UNAVAILABLE}
    interceptor = RetryOnRpcErrorClientInterceptor(
        DummyRetryConfig(retryable_status=retryable_codes)
    )
    not_errors = [object() for _ in range(1000)]
    for obj in not_errors:
        codeflash_output = interceptor._is_retryable_error(obj) # 183μs -> 197μs (7.15% slower)

def test_large_mixed_error_types_and_codes():
    """Test a large mixed list of error objects with varying retryable status and class names."""
    retryable_codes = {grpc.StatusCode.UNAVAILABLE, grpc.StatusCode.DEADLINE_EXCEEDED}
    interceptor = RetryOnRpcErrorClientInterceptor(
        DummyRetryConfig(retryable_status=retryable_codes)
    )
    errors = []
    # 250 retryable DummyRpcError
    errors += [DummyRpcError(grpc.StatusCode.UNAVAILABLE) for _ in range(250)]
    # 250 non-retryable DummyRpcError
    errors += [DummyRpcError(grpc.StatusCode.PERMISSION_DENIED) for _ in range(250)]
    # 250 DummyMultiThreadedRendezvous (should never be retryable)
    errors += [DummyMultiThreadedRendezvous(grpc.StatusCode.UNAVAILABLE) for _ in range(250)]
    # 250 objects not RpcError
    errors += [object() for _ in range(250)]
    # Check
    for i, err in enumerate(errors):
        if i < 250:
            codeflash_output = interceptor._is_retryable_error(err)
        else:
            codeflash_output = interceptor._is_retryable_error(err)
# codeflash_output is used to check that the output of the original code is the same as that of the optimized code.
#------------------------------------------------
import grpc
# imports
import pytest
from pinecone.grpc.retry import RetryOnRpcErrorClientInterceptor

# --- Helper classes for testing ---

class DummyRetryConfig:
    """Dummy config for RetryOnRpcErrorClientInterceptor."""
    def __init__(self, retryable_status):
        self.max_attempts = 3
        self.sleep_policy = lambda x: x
        self.retryable_status = retryable_status

class DummyRpcError(grpc.RpcError):
    """Minimal RpcError for testing."""
    def __init__(self, code_value):
        self._code_value = code_value

    def code(self):
        return self._code_value

class DummyNonRpcError(Exception):
    """Non-RpcError for negative tests."""
    pass

class DummyMultiThreadedRendezvous(grpc.RpcError):
    """Simulate _MultiThreadedRendezvous error."""
    def __init__(self, code_value):
        self._code_value = code_value

    def code(self):
        return self._code_value

    @property
    def __class__(self):
        # Simulate class name containing "_MultiThreadedRendezvous"
        class Dummy:
            __name__ = "_MultiThreadedRendezvous"
        return Dummy

# --- Basic Test Cases ---
def test_basic_retryable_error_true():
    """Should return True for retryable RpcError with correct code."""
    retryable_status = {grpc.StatusCode.UNAVAILABLE, grpc.StatusCode.DEADLINE_EXCEEDED}
    interceptor = RetryOnRpcErrorClientInterceptor(DummyRetryConfig(retryable_status))
    error = DummyRpcError(grpc.StatusCode.UNAVAILABLE)
    codeflash_output = interceptor._is_retryable_error(error) # 1.53μs -> 1.43μs (6.77% faster)

def test_basic_retryable_error_false_wrong_code():
    """Should return False for RpcError with non-retryable code."""
    retryable_status = {grpc.StatusCode.UNAVAILABLE}
    interceptor = RetryOnRpcErrorClientInterceptor(DummyRetryConfig(retryable_status))
    error = DummyRpcError(grpc.StatusCode.INVALID_ARGUMENT)
    codeflash_output = interceptor._is_retryable_error(error) # 1.44μs -> 1.41μs (1.99% faster)

def test_basic_non_rpc_error():
    """Should return False for non-RpcError instances."""
    retryable_status = {grpc.StatusCode.UNAVAILABLE}
    interceptor = RetryOnRpcErrorClientInterceptor(DummyRetryConfig(retryable_status))
    error = DummyNonRpcError("Not an RpcError")
    codeflash_output = interceptor._is_retryable_error(error) # 799ns -> 735ns (8.71% faster)

# --- Edge Test Cases ---
def test_edge_multithreaded_rendezvous_class_name():
    """Should return False for RpcError with class name containing '_MultiThreadedRendezvous'."""
    retryable_status = {grpc.StatusCode.UNAVAILABLE}
    interceptor = RetryOnRpcErrorClientInterceptor(DummyRetryConfig(retryable_status))
    error = DummyMultiThreadedRendezvous(grpc.StatusCode.UNAVAILABLE)
    codeflash_output = interceptor._is_retryable_error(error) # 12.7μs -> 1.63μs (681% faster)

def test_edge_empty_retryable_status():
    """Should return False when retryable_status set is empty."""
    retryable_status = set()
    interceptor = RetryOnRpcErrorClientInterceptor(DummyRetryConfig(retryable_status))
    error = DummyRpcError(grpc.StatusCode.UNAVAILABLE)
    codeflash_output = interceptor._is_retryable_error(error) # 1.63μs -> 1.60μs (1.81% faster)

def test_edge_none_input():
    """Should return False when input is None."""
    retryable_status = {grpc.StatusCode.UNAVAILABLE}
    interceptor = RetryOnRpcErrorClientInterceptor(DummyRetryConfig(retryable_status))
    codeflash_output = interceptor._is_retryable_error(None) # 566ns -> 572ns (1.05% slower)

def test_edge_code_method_raises():
    """Should handle RpcError whose code() method raises an exception."""
    class BadRpcError(grpc.RpcError):
        def code(self):
            raise RuntimeError("code() failed")
    retryable_status = {grpc.StatusCode.UNAVAILABLE}
    interceptor = RetryOnRpcErrorClientInterceptor(DummyRetryConfig(retryable_status))
    error = BadRpcError()
    with pytest.raises(RuntimeError):
        # The function does not catch exceptions from code()
        interceptor._is_retryable_error(error) # 1.96μs -> 1.70μs (15.4% faster)

def test_edge_retryable_status_contains_none():
    """Should return False if code() returns None and retryable_status contains None."""
    retryable_status = {None}
    interceptor = RetryOnRpcErrorClientInterceptor(DummyRetryConfig(retryable_status))
    class NoneCodeRpcError(grpc.RpcError):
        def code(self):
            return None
    error = NoneCodeRpcError()
    codeflash_output = interceptor._is_retryable_error(error) # 1.43μs -> 1.22μs (17.4% faster)

def test_edge_retryable_status_contains_multiple_types():
    """Should handle retryable_status with mixed types gracefully."""
    retryable_status = {grpc.StatusCode.UNAVAILABLE, 123, "foo"}
    interceptor = RetryOnRpcErrorClientInterceptor(DummyRetryConfig(retryable_status))
    error = DummyRpcError(123)
    codeflash_output = interceptor._is_retryable_error(error) # 1.26μs -> 1.21μs (4.56% faster)
    error2 = DummyRpcError("foo")
    codeflash_output = interceptor._is_retryable_error(error2) # 501ns -> 516ns (2.91% slower)
    error3 = DummyRpcError(grpc.StatusCode.UNAVAILABLE)
    codeflash_output = interceptor._is_retryable_error(error3) # 559ns -> 587ns (4.77% slower)
    error4 = DummyRpcError(grpc.StatusCode.INVALID_ARGUMENT)
    codeflash_output = interceptor._is_retryable_error(error4) # 528ns -> 576ns (8.33% slower)

# --- Large Scale Test Cases ---
def test_large_scale_many_status_codes():
    """Should return True for RpcError with code in a large retryable_status set."""
    # Use 100 status codes (simulate with integers)
    retryable_status = set(range(100))
    interceptor = RetryOnRpcErrorClientInterceptor(DummyRetryConfig(retryable_status))
    # Test for all codes in the set
    for code in range(100):
        error = DummyRpcError(code)
        codeflash_output = interceptor._is_retryable_error(error) # 31.7μs -> 32.6μs (2.59% slower)
    # Test for codes outside the set
    for code in range(100, 110):
        error = DummyRpcError(code)
        codeflash_output = interceptor._is_retryable_error(error) # 3.05μs -> 3.18μs (4.21% slower)

To edit these changes git checkout codeflash/optimize-RetryOnRpcErrorClientInterceptor._is_retryable_error-mh9h2032 and push.