Communication in Distributed Systems

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

// Protocol Buffer definition for a Calculator service
syntax = "proto3";
 
package calculator;
 
// Message definitions
message AddRequest {
    int32 a = 1;        // Field number 1
    int32 b = 2;        // Field number 2
    string request_id = 3;  // Optional, for tracing
}
 
message AddResponse {
    int32 result = 1;
    int64 compute_time_ns = 2;  // Processing metadata
}
 
message UserContext {
    string user_id = 1;
    repeated string roles = 2;     // Array of roles
    map<string, string> metadata = 3;  // Key-value pairs
}
 
// Complex nested message
message BatchRequest {
    UserContext context = 1;
    repeated AddRequest requests = 2;
    
    enum Priority {
        LOW = 0;
        NORMAL = 1;
        HIGH = 2;
    }
    Priority priority = 3;
}
 
message BatchResponse {
    repeated AddResponse results = 1;
    int32 total_compute_time_ms = 2;
}
 
// Service definition (for gRPC)
service Calculator {
    rpc Add(AddRequest) returns (AddResponse);
    rpc BatchAdd(BatchRequest) returns (BatchResponse);
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79

# Protocol Buffers marshaling/unmarshaling in Python
# Generated from calculator.proto using: protoc --python_out=. calculator.proto
 
from calculator_pb2 import AddRequest, AddResponse
 
def demonstrate_marshaling():
    # Create a message
    request = AddRequest()
    request.a = 5
    request.b = 3
    request.request_id = "req-12345"
    
    # Marshal to bytes
    wire_bytes = request.SerializeToString()
    print(f"Serialized size: {len(wire_bytes)} bytes")
    print(f"Wire format: {wire_bytes.hex()}")
    
    # Output:
    # Serialized size: 15 bytes
    # Wire format: 0805100318097265712d3132333435
    
    # Unmarshal from bytes
    received = AddRequest()
    received.ParseFromString(wire_bytes)
    
    print(f"Received: a={received.a}, b={received.b}, id={received.request_id}")
 
 
def demonstrate_forward_compatibility():
    """
    Protobuf handles version differences gracefully.
    Old code can read messages with new fields (ignores them).
    New code can read messages missing optional fields (uses defaults).
    """
    
    # Imagine we're an old client that doesn't know about request_id
    old_format_bytes = bytes.fromhex("08051003")  # Just a and b
    
    # New code can still parse it (request_id will be empty string)
    request = AddRequest()
    request.ParseFromString(old_format_bytes)
    
    print(f"a={request.a}, b={request.b}")  # Works fine
    print(f"request_id='{request.request_id}'")  # Empty string (default)
    
    # Conversely, new messages work with old code
    # Old code will ignore unknown field 3 (request_id)
 
 
def benchmark_serialization():
    """Compare serialization formats"""
    import json
    import time
    
    # Create test data
    request = AddRequest(a=5, b=3, request_id="req-12345")
    json_data = {"a": 5, "b": 3, "request_id": "req-12345"}
    
    iterations = 100000
    
    # Benchmark protobuf
    start = time.time()
    for _ in range(iterations):
        wire = request.SerializeToString()
        parsed = AddRequest()
        parsed.ParseFromString(wire)
    proto_time = time.time() - start
    
    # Benchmark JSON
    start = time.time()
    for _ in range(iterations):
        wire = json.dumps(json_data).encode('utf-8')
        parsed = json.loads(wire.decode('utf-8'))
    json_time = time.time() - start
    
    print(f"Protobuf: {proto_time:.2f}s ({len(request.SerializeToString())} bytes)")
    print(f"JSON: {json_time:.2f}s ({len(json.dumps(json_data))} bytes)")
    print(f"Protobuf is {json_time/proto_time:.1f}x faster, "
          f"{len(json.dumps(json_data))/len(request.SerializeToString()):.1f}x smaller")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146

# Service registration and discovery with Consul
import consul
import socket
import time
from threading import Thread
 
class ServiceRegistry:
    """
    Wrapper around Consul for service registration and discovery.
    Enables location-transparent RPC.
    """
    
    def __init__(self, consul_host='localhost', consul_port=8500):
        self.consul = consul.Consul(host=consul_host, port=consul_port)
        self.registered_services = []
    
    def register(self, service_name, port, tags=None, health_check_interval='10s'):
        """
        Register this service instance with Consul.
        Includes health check so Consul knows when we're down.
        """
        service_id = f"{service_name}-{socket.gethostname()}-{port}"
        
        # Register service with health check
        self.consul.agent.service.register(
            name=service_name,
            service_id=service_id,
            port=port,
            tags=tags or [],
            check=consul.Check.tcp(
                host='localhost',
                port=port,
                interval=health_check_interval,
                timeout='5s',
                deregister='1m',  # Remove if unhealthy for 1 minute
            )
        )
        
        self.registered_services.append(service_id)
        print(f"Registered {service_id}")
    
    def discover(self, service_name, healthy_only=True):
        """
        Discover all instances of a service.
        Returns list of (host, port) tuples.
        """
        _, services = self.consul.health.service(
            service_name, 
            passing=healthy_only
        )
        
        instances = []
        for service in services:
            host = service['Service']['Address'] or service['Node']['Address']
            port = service['Service']['Port']
            instances.append((host, port))
        
        return instances
    
    def discover_one(self, service_name, strategy='random'):
        """
        Discover a single instance (for load balancing).
        """
        import random
        instances = self.discover(service_name)
        
        if not instances:
            raise ServiceNotFoundError(f"No healthy instances of {service_name}")
        
        if strategy == 'random':
            return random.choice(instances)
        elif strategy == 'first':
            return instances[0]
        # Could implement round-robin, least-connections, etc.
    
    def deregister_all(self):
        """Deregister all services on shutdown"""
        for service_id in self.registered_services:
            self.consul.agent.service.deregister(service_id)
 
 
class RpcClient:
    """
    RPC client with automatic service discovery.
    """
    
    def __init__(self, registry: ServiceRegistry):
        self.registry = registry
        self.connections = {}  # Cache connections
    
    def call(self, service_name, method, *args, **kwargs):
        """
        Call a method on a named service.
        Handles discovery and connection management.
        """
        # Discover a healthy instance
        host, port = self.registry.discover_one(service_name)
        
        # Get or create connection
        conn = self._get_connection(host, port)
        
        try:
            return conn.call(method, *args, **kwargs)
        except ConnectionError:
            # Connection failed - remove from cache, retry with different instance
            self._remove_connection(host, port)
            return self.call(service_name, method, *args, **kwargs)
    
    def _get_connection(self, host, port):
        key = (host, port)
        if key not in self.connections:
            self.connections[key] = create_rpc_connection(host, port)
        return self.connections[key]
 
 
# Server-side usage
def run_server():
    registry = ServiceRegistry()
    
    # Start server on a port
    server = CalculatorServer(port=50051)
    server_thread = Thread(target=server.serve)
    server_thread.start()
    
    # Register with Consul
    registry.register(
        'calculator-service',
        port=50051,
        tags=['v1', 'production'],
    )
    
    # On shutdown
    try:
        server_thread.join()
    finally:
        registry.deregister_all()
 
 
# Client-side usage  
def client_example():
    registry = ServiceRegistry()
    client = RpcClient(registry)
    
    # Call by service name - no hardcoded addresses!
    result = client.call('calculator-service', 'add', a=5, b=3)
    print(f"Result: {result}")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139

# RPC call semantics implementations
 
import uuid
import time
from abc import ABC, abstractmethod
 
class RpcSemantics(ABC):
    """Base class for different RPC call semantics"""
    
    @abstractmethod
    def call(self, stub, method, *args, **kwargs):
        pass
 
 
class MaybeSemantics(RpcSemantics):
    """
    Maybe semantics: Try once, return error on failure.
    Execution: 0 or 1 times
    """
    
    def call(self, stub, method, *args, timeout=5.0, **kwargs):
        try:
            return stub.call_with_timeout(method, args, kwargs, timeout)
        except TimeoutError:
            raise RpcError("Call failed - no retry attempted")
 
 
class AtLeastOnceSemantics(RpcSemantics):
    """
    At-least-once semantics: Retry until success.
    Execution: 1 or more times
    Warning: Only safe for idempotent operations!
    """
    
    def __init__(self, max_retries=5, base_delay=1.0):
        self.max_retries = max_retries
        self.base_delay = base_delay
    
    def call(self, stub, method, *args, timeout=5.0, **kwargs):
        last_error = None
        
        for attempt in range(self.max_retries):
            try:
                return stub.call_with_timeout(method, args, kwargs, timeout)
            except TimeoutError as e:
                last_error = e
                delay = self.base_delay * (2 ** attempt)
                time.sleep(delay)
        
        raise RpcError(f"Call failed after {self.max_retries} retries") from last_error
 
 
class AtMostOnceSemantics(RpcSemantics):
    """
    At-most-once semantics: Try once, no automatic retry.
    Execution: 0 or 1 times
    Client can manually retry but must handle duplicates.
    """
    
    def call(self, stub, method, *args, timeout=5.0, **kwargs):
        return MaybeSemantics().call(stub, method, *args, timeout=timeout, **kwargs)
 
 
class ExactlyOnceSemantics(RpcSemantics):
    """
    Exactly-once semantics: Request is executed exactly once.
    Requires:
    1. Unique request ID per logical request
    2. Server-side request deduplication
    3. Persistent storage of processed IDs
    4. Client retries with same request ID
    """
    
    def __init__(self, max_retries=5, base_delay=1.0):
        self.max_retries = max_retries
        self.base_delay = base_delay
    
    def call(self, stub, method, *args, timeout=5.0, **kwargs):
        # Generate unique ID for this logical request
        # Same ID must be used across all retries
        request_id = kwargs.pop('_request_id', str(uuid.uuid4()))
        
        for attempt in range(self.max_retries):
            try:
                return stub.call_with_request_id(
                    method, args, kwargs, 
                    request_id=request_id,
                    timeout=timeout
                )
            except TimeoutError:
                delay = self.base_delay * (2 ** attempt)
                time.sleep(delay)
        
        raise RpcError(f"Call failed after {self.max_retries} retries")
 
 
class ExactlyOnceServer:
    """
    Server-side support for exactly-once semantics.
    Tracks processed request IDs to detect and handle duplicates.
    """
    
    def __init__(self, persistence):
        self.persistence = persistence  # Must survive crashes!
    
    def handle_request(self, request_id, procedure, args, kwargs):
        # Check if we've already processed this request
        cached_result = self.persistence.get_result(request_id)
        if cached_result is not None:
            # Duplicate! Return cached result without re-executing
            return cached_result
        
        # First time seeing this request - execute
        result = self.execute(procedure, args, kwargs)
        
        # Persist result before responding (crash tolerance)
        self.persistence.store_result(request_id, result)
        
        return result
    
    def execute(self, procedure, args, kwargs):
        """Execute the actual procedure"""
        return procedure(*args, **kwargs)
 
 
# Example usage
def example_client():
    client = RpcClient(AtLeastOnceSemantics(max_retries=3))
    
    # Safe: idempotent operation
    balance = client.call(bank_service, 'get_balance', account_id='12345')
    
    # Dangerous with at-least-once!
    # client.call(bank_service, 'transfer', amount=100, from_='A', to='B')
    
    # Safe: use exactly-once for non-idempotent operations
    tx_client = RpcClient(ExactlyOnceSemantics())
    tx_client.call(bank_service, 'transfer', amount=100, from_='A', to='B',
                   _request_id='user-generated-tx-id-12345')

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106

# Patterns for handling RPC-specific challenges
 
# 1. BATCH OPERATIONS
# Instead of N round-trips, use one batch call
 
# Bad: N remote calls
def process_users_bad(user_ids):
    results = []
    for user_id in user_ids:
        user = user_service.get_user(user_id)  # Remote call each time!
        results.append(user)
    return results
 
# Good: One batch call
def process_users_good(user_ids):
    return user_service.get_users_batch(user_ids)  # Single remote call
 
 
# 2. ASYNCHRONOUS/PARALLEL CALLS
# When calls are independent, run them concurrently
 
import asyncio
 
async def fetch_dashboard_data(user_id):
    # Three independent RPC calls
    # Sequential: 50ms + 50ms + 50ms = 150ms
    # Parallel:   max(50ms, 50ms, 50ms) = 50ms
    
    user_task = asyncio.create_task(user_service.get_profile(user_id))
    orders_task = asyncio.create_task(order_service.get_recent(user_id))
    recs_task = asyncio.create_task(rec_service.get_recommendations(user_id))
    
    user, orders, recommendations = await asyncio.gather(
        user_task, orders_task, recs_task
    )
    
    return DashboardData(user, orders, recommendations)
 
 
# 3. HANDLING PARTIAL FAILURE
# Use defensive programming and explicit failure handling
 
class UserServiceClient:
    def __init__(self, timeout=5.0, fallback_mode=True):
        self.timeout = timeout
        self.fallback_mode = fallback_mode
        self.cache = LocalCache()
    
    async def get_user(self, user_id):
        try:
            user = await asyncio.wait_for(
                self.stub.get_user(user_id),
                timeout=self.timeout
            )
            self.cache.set(user_id, user)
            return user
            
        except asyncio.TimeoutError:
            # Timeout - RPC didn't complete in time
            # Could be: network issue, server slow, server dead
            if self.fallback_mode:
                cached = self.cache.get(user_id)
                if cached:
                    return CachedUser(cached, stale=True)
            raise ServiceUnavailableError("User service timeout")
            
        except ConnectionError:
            # Connection failed - server definitely unreachable
            raise ServiceUnavailableError("User service unreachable")
            
        except RpcError as e:
            # Server returned an error
            if e.code == ErrorCode.NOT_FOUND:
                raise UserNotFoundError(user_id)
            raise  # Re-raise unexpected errors
 
 
# 4. TIMEOUTS AND DEADLINES
# Always set timeouts! Propagate deadlines through call chains.
 
class DeadlineContext:
    """Propagate deadline through RPC call chain"""
    
    def __init__(self, deadline_seconds):
        self.deadline = time.time() + deadline_seconds
    
    def remaining(self):
        return max(0, self.deadline - time.time())
    
    def is_expired(self):
        return time.time() > self.deadline
 
 
def api_handler(request):
    # API has 30s timeout
    ctx = DeadlineContext(30)
    
    # Each service call gets remaining time, minus buffer
    user = user_service.get(request.user_id, timeout=ctx.remaining() - 1)
    
    if ctx.is_expired():
        raise TimeoutError("Deadline exceeded")
    
    orders = order_service.get_for_user(request.user_id, timeout=ctx.remaining() - 1)
    
    return Response(user, orders)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

// gRPC service definition
syntax = "proto3";
 
package calculator;
 
option go_package = "github.com/example/calculator";
 
// Request/Response messages
message Number {
    double value = 1;
}
 
message BinaryOpRequest {
    double a = 1;
    double b = 2;
}
 
message NumberStream {
    repeated double values = 1;
}
 
message Statistics {
    double sum = 1;
    double average = 2;
    double min = 3;
    double max = 4;
    int64 count = 5;
}
 
// The Calculator service with all RPC patterns
service Calculator {
    // Unary: one request, one response
    rpc Add(BinaryOpRequest) returns (Number);
    rpc Multiply(BinaryOpRequest) returns (Number);
    
    // Server streaming: one request, stream of responses
    rpc GenerateSequence(Number) returns (stream Number);
    
    // Client streaming: stream of requests, one response
    rpc ComputeStatistics(stream Number) returns (Statistics);
    
    // Bidirectional streaming: both sides stream
    rpc StreamingCalculator(stream BinaryOpRequest) returns (stream Number);
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147

# gRPC server and client implementation
# Generated from: python -m grpc_tools.protoc -I. --python_out=. --grpc_python_out=. calculator.proto
 
import grpc
from concurrent import futures
import calculator_pb2
import calculator_pb2_grpc
 
# =================== SERVER IMPLEMENTATION ===================
 
class CalculatorServicer(calculator_pb2_grpc.CalculatorServicer):
    """gRPC server implementation"""
    
    def Add(self, request, context):
        """Unary RPC: single request, single response"""
        result = request.a + request.b
        return calculator_pb2.Number(value=result)
    
    def GenerateSequence(self, request, context):
        """Server streaming: client sends one request, server sends multiple responses"""
        n = int(request.value)
        for i in range(n):
            yield calculator_pb2.Number(value=float(i))
    
    def ComputeStatistics(self, request_iterator, context):
        """Client streaming: client streams values, server returns single summary"""
        values = []
        for request in request_iterator:
            values.append(request.value)
        
        if not values:
            return calculator_pb2.Statistics()
        
        return calculator_pb2.Statistics(
            sum=sum(values),
            average=sum(values) / len(values),
            min=min(values),
            max=max(values),
            count=len(values),
        )
    
    def StreamingCalculator(self, request_iterator, context):
        """Bidirectional streaming: both sides stream simultaneously"""
        for request in request_iterator:
            result = request.a + request.b
            yield calculator_pb2.Number(value=result)
 
 
def serve():
    """Start gRPC server"""
    server = grpc.server(
        futures.ThreadPoolExecutor(max_workers=10),
        options=[
            ('grpc.max_receive_message_length', 10 * 1024 * 1024),
            ('grpc.keepalive_time_ms', 30000),
        ]
    )
    calculator_pb2_grpc.add_CalculatorServicer_to_server(
        CalculatorServicer(), server
    )
    server.add_insecure_port('[::]:50051')
    server.start()
    print("Server started on port 50051")
    server.wait_for_termination()
 
 
# =================== CLIENT IMPLEMENTATION ===================
 
def create_client():
    """Create gRPC client with interceptors"""
    channel = grpc.insecure_channel(
        'localhost:50051',
        options=[
            ('grpc.keepalive_time_ms', 30000),
            ('grpc.keepalive_timeout_ms', 5000),
        ]
    )
    return calculator_pb2_grpc.CalculatorStub(channel)
 
 
def unary_example(stub):
    """Simple request-response call"""
    request = calculator_pb2.BinaryOpRequest(a=5.0, b=3.0)
    
    # With timeout/deadline
    response = stub.Add(request, timeout=5.0)
    print(f"5 + 3 = {response.value}")
 
 
def server_streaming_example(stub):
    """Server streams multiple responses"""
    request = calculator_pb2.Number(value=10.0)
    
    # Returns an iterator of responses
    for response in stub.GenerateSequence(request):
        print(f"Received: {response.value}")
 
 
def client_streaming_example(stub):
    """Client streams multiple requests"""
    def generate_numbers():
        for i in range(100):
            yield calculator_pb2.Number(value=float(i))
    
    response = stub.ComputeStatistics(generate_numbers())
    print(f"Statistics: sum={response.sum}, avg={response.average}")
 
 
def bidirectional_example(stub):
    """Both sides stream simultaneously"""
    def generate_requests():
        pairs = [(1, 2), (3, 4), (5, 6), (7, 8)]
        for a, b in pairs:
            yield calculator_pb2.BinaryOpRequest(a=float(a), b=float(b))
    
    # Call returns an iterator, we stream requests while getting responses
    responses = stub.StreamingCalculator(generate_requests())
    
    for response in responses:
        print(f"Result: {response.value}")
 
 
# =================== METADATA AND DEADLINES ===================
 
def call_with_metadata(stub):
    """Pass metadata (like HTTP headers)"""
    metadata = [
        ('authorization', 'Bearer token123'),
        ('x-request-id', 'req-abc-123'),
        ('x-correlation-id', 'corr-xyz-789'),
    ]
    
    request = calculator_pb2.BinaryOpRequest(a=5.0, b=3.0)
    response = stub.Add(request, metadata=metadata)
 
 
def call_with_deadline(stub):
    """Set deadline that propagates through call chain"""
    request = calculator_pb2.BinaryOpRequest(a=5.0, b=3.0)
    
    try:
        # Deadline: call must complete within 100ms
        response = stub.Add(request, timeout=0.1)
    except grpc.RpcError as e:
        if e.code() == grpc.StatusCode.DEADLINE_EXCEEDED:
            print("Call timed out!")
        raise

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153

# gRPC interceptors for cross-cutting concerns
 
import grpc
import time
import logging
from typing import Callable, Any
 
# =================== CLIENT INTERCEPTORS ===================
 
class LoggingClientInterceptor(grpc.UnaryUnaryClientInterceptor):
    """Log all outgoing RPC calls"""
    
    def intercept_unary_unary(self, continuation, client_call_details, request):
        method = client_call_details.method
        logging.info(f"Calling {method}")
        
        start_time = time.time()
        response = continuation(client_call_details, request)
        latency_ms = (time.time() - start_time) * 1000
        
        logging.info(f"{method} completed in {latency_ms:.1f}ms")
        return response
 
 
class RetryClientInterceptor(grpc.UnaryUnaryClientInterceptor):
    """Automatically retry failed calls"""
    
    def __init__(self, max_retries=3, retryable_codes=None):
        self.max_retries = max_retries
        self.retryable_codes = retryable_codes or {
            grpc.StatusCode.UNAVAILABLE,
            grpc.StatusCode.DEADLINE_EXCEEDED,
        }
    
    def intercept_unary_unary(self, continuation, client_call_details, request):
        for attempt in range(self.max_retries + 1):
            try:
                return continuation(client_call_details, request)
            except grpc.RpcError as e:
                if e.code() not in self.retryable_codes:
                    raise  # Non-retryable error
                if attempt == self.max_retries:
                    raise  # Max retries exceeded
                
                delay = (2 ** attempt) * 0.1  # Exponential backoff
                time.sleep(delay)
 
 
class TracingClientInterceptor(grpc.UnaryUnaryClientInterceptor):
    """Inject distributed trace context into calls"""
    
    def intercept_unary_unary(self, continuation, client_call_details, request):
        # Get current trace context (e.g., from OpenTelemetry)
        trace_id = get_current_trace_id()
        span_id = create_new_span_id()
        
        # Add trace headers to metadata
        new_metadata = list(client_call_details.metadata or [])
        new_metadata.extend([
            ('x-trace-id', trace_id),
            ('x-span-id', span_id),
            ('x-parent-span-id', get_current_span_id()),
        ])
        
        new_details = client_call_details._replace(metadata=new_metadata)
        return continuation(new_details, request)
 
 
# =================== SERVER INTERCEPTORS ===================
 
class AuthenticationServerInterceptor(grpc.ServerInterceptor):
    """Validate authentication before handling requests"""
    
    def __init__(self, validator):
        self.validator = validator
    
    def intercept_service(self, continuation, handler_call_details):
        method = handler_call_details.method
        
        # Skip auth for health checks
        if method == '/grpc.health.v1.Health/Check':
            return continuation(handler_call_details)
        
        # Extract token from metadata
        metadata = dict(handler_call_details.invocation_metadata)
        token = metadata.get('authorization')
        
        if not token:
            return self._abort(grpc.StatusCode.UNAUTHENTICATED, "Missing token")
        
        try:
            user = self.validator.validate(token)
            # Attach user to context for handler use
            handler_call_details.user = user
            return continuation(handler_call_details)
        except InvalidTokenError:
            return self._abort(grpc.StatusCode.UNAUTHENTICATED, "Invalid token")
    
    def _abort(self, code, message):
        def abort_handler(request, context):
            context.abort(code, message)
        return grpc.unary_unary_rpc_method_handler(abort_handler)
 
 
class MetricsServerInterceptor(grpc.ServerInterceptor):
    """Collect request metrics"""
    
    def __init__(self, metrics_client):
        self.metrics = metrics_client
    
    def intercept_service(self, continuation, handler_call_details):
        method = handler_call_details.method
        start_time = time.time()
        
        try:
            response = continuation(handler_call_details)
            self.metrics.increment(f'rpc.{method}.success')
            return response
        except grpc.RpcError as e:
            self.metrics.increment(f'rpc.{method}.error.{e.code().name}')
            raise
        finally:
            latency_ms = (time.time() - start_time) * 1000
            self.metrics.histogram(f'rpc.{method}.latency_ms', latency_ms)
 
 
# =================== WIRING INTERCEPTORS ===================
 
def create_client_with_interceptors():
    """Create client channel with interceptor chain"""
    channel = grpc.insecure_channel('localhost:50051')
    
    # Interceptors execute in order: logging -> retry -> tracing
    channel = grpc.intercept_channel(
        channel,
        LoggingClientInterceptor(),
        RetryClientInterceptor(max_retries=3),
        TracingClientInterceptor(),
    )
    
    return calculator_pb2_grpc.CalculatorStub(channel)
 
 
def create_server_with_interceptors():
    """Create server with interceptor chain"""
    server = grpc.server(
        futures.ThreadPoolExecutor(max_workers=10),
        interceptors=[
            AuthenticationServerInterceptor(TokenValidator()),
            MetricsServerInterceptor(PrometheusClient()),
        ]
    )
    return server