Synchronous Communication

Introduction

Synchronous communication is the most intuitive way for services to interact—one service calls another and waits for a response. Through building systems with dozens of interconnected services, I've learned that while synchronous calls are simple to understand, they introduce coupling and cascading failure risks that require careful management.

This article covers HTTP/REST communication patterns, gRPC for high-performance scenarios, service discovery, and load balancing strategies.

Communication Patterns Overview

Pattern	Use Case	Trade-offs
Synchronous	Need immediate response	Simple but creates coupling
Asynchronous	Fire-and-forget, eventual consistency	Complex but decoupled

HTTP/REST Communication

Basic HTTP Client

import httpx
from typing import TypeVar, Generic
from pydantic import BaseModel
from dataclasses import dataclass

T = TypeVar("T")


@dataclass
class ServiceConfig:
    base_url: str
    timeout: float = 30.0
    max_retries: int = 3


class HTTPServiceClient:
    """Base HTTP client for service-to-service communication."""
    
    def __init__(self, config: ServiceConfig):
        self.config = config
        self.client = httpx.AsyncClient(
            base_url=config.base_url,
            timeout=httpx.Timeout(config.timeout),
        )
    
    async def get(self, path: str, params: dict | None = None) -> dict:
        response = await self.client.get(path, params=params)
        response.raise_for_status()
        return response.json()
    
    async def post(self, path: str, data: dict) -> dict:
        response = await self.client.post(path, json=data)
        response.raise_for_status()
        return response.json()
    
    async def put(self, path: str, data: dict) -> dict:
        response = await self.client.put(path, json=data)
        response.raise_for_status()
        return response.json()
    
    async def delete(self, path: str) -> None:
        response = await self.client.delete(path)
        response.raise_for_status()
    
    async def close(self):
        await self.client.aclose()


# Usage
class UserServiceClient(HTTPServiceClient):
    """Client for User Service."""
    
    async def get_user(self, user_id: str) -> dict:
        return await self.get(f"/users/{user_id}")
    
    async def create_user(self, email: str, name: str) -> dict:
        return await self.post("/users", {"email": email, "name": name})


# In Order Service
user_client = UserServiceClient(
    ServiceConfig(base_url="http://user-service:8000")
)
user = await user_client.get_user("123")

Request/Response Correlation

import uuid
from contextvars import ContextVar
from fastapi import Request, Response
from starlette.middleware.base import BaseHTTPMiddleware

# Context variable for request tracing
request_id_var: ContextVar[str] = ContextVar("request_id", default="")
correlation_id_var: ContextVar[str] = ContextVar("correlation_id", default="")


class TracingMiddleware(BaseHTTPMiddleware):
    """Propagate tracing headers across service calls."""
    
    async def dispatch(self, request: Request, call_next):
        # Extract or generate request ID
        request_id = request.headers.get("X-Request-ID", str(uuid.uuid4()))
        correlation_id = request.headers.get(
            "X-Correlation-ID", 
            request_id  # Use request ID as correlation ID if not provided
        )
        
        # Set context variables
        request_id_var.set(request_id)
        correlation_id_var.set(correlation_id)
        
        # Process request
        response: Response = await call_next(request)
        
        # Add headers to response
        response.headers["X-Request-ID"] = request_id
        response.headers["X-Correlation-ID"] = correlation_id
        
        return response


class TracingHTTPClient(HTTPServiceClient):
    """HTTP client that propagates tracing headers."""
    
    async def _request(self, method: str, path: str, **kwargs) -> httpx.Response:
        headers = kwargs.pop("headers", {})
        
        # Add tracing headers
        headers["X-Request-ID"] = str(uuid.uuid4())
        headers["X-Correlation-ID"] = correlation_id_var.get()
        
        return await self.client.request(method, path, headers=headers, **kwargs)
    
    async def get(self, path: str, params: dict | None = None) -> dict:
        response = await self._request("GET", path, params=params)
        response.raise_for_status()
        return response.json()

Error Handling in Service Calls

from enum import Enum
from dataclasses import dataclass
import httpx


class ServiceErrorType(Enum):
    NOT_FOUND = "not_found"
    VALIDATION_ERROR = "validation_error"
    UNAUTHORIZED = "unauthorized"
    SERVICE_UNAVAILABLE = "service_unavailable"
    UNKNOWN = "unknown"


@dataclass
class ServiceError(Exception):
    error_type: ServiceErrorType
    message: str
    service: str
    status_code: int | None = None
    details: dict | None = None


class ResilientServiceClient(HTTPServiceClient):
    """Service client with proper error handling."""
    
    def __init__(self, config: ServiceConfig, service_name: str):
        super().__init__(config)
        self.service_name = service_name
    
    def _handle_error(self, error: httpx.HTTPStatusError) -> ServiceError:
        """Convert HTTP errors to domain errors."""
        status = error.response.status_code
        
        try:
            body = error.response.json()
            message = body.get("message", str(error))
            details = body.get("details")
        except Exception:
            message = str(error)
            details = None
        
        if status == 404:
            error_type = ServiceErrorType.NOT_FOUND
        elif status == 422:
            error_type = ServiceErrorType.VALIDATION_ERROR
        elif status == 401 or status == 403:
            error_type = ServiceErrorType.UNAUTHORIZED
        elif status >= 500:
            error_type = ServiceErrorType.SERVICE_UNAVAILABLE
        else:
            error_type = ServiceErrorType.UNKNOWN
        
        return ServiceError(
            error_type=error_type,
            message=message,
            service=self.service_name,
            status_code=status,
            details=details,
        )
    
    async def get(self, path: str, params: dict | None = None) -> dict:
        try:
            response = await self.client.get(path, params=params)
            response.raise_for_status()
            return response.json()
        except httpx.HTTPStatusError as e:
            raise self._handle_error(e)
        except httpx.ConnectError:
            raise ServiceError(
                error_type=ServiceErrorType.SERVICE_UNAVAILABLE,
                message=f"Cannot connect to {self.service_name}",
                service=self.service_name,
            )


# Usage in Order Service
class OrderService:
    def __init__(self, user_client: ResilientServiceClient):
        self.user_client = user_client
    
    async def create_order(self, user_id: str, items: list) -> dict:
        try:
            user = await self.user_client.get(f"/users/{user_id}")
        except ServiceError as e:
            if e.error_type == ServiceErrorType.NOT_FOUND:
                raise ValueError(f"User {user_id} not found")
            elif e.error_type == ServiceErrorType.SERVICE_UNAVAILABLE:
                # Could queue for retry or return degraded response
                raise
            raise
        
        # Continue with order creation
        return {"user": user, "items": items}

gRPC for High-Performance Communication

Protocol Buffers Definition

// proto/user_service.proto
syntax = "proto3";

package user;

service UserService {
    rpc GetUser(GetUserRequest) returns (User);
    rpc CreateUser(CreateUserRequest) returns (User);
    rpc ListUsers(ListUsersRequest) returns (ListUsersResponse);
    rpc StreamUserUpdates(StreamRequest) returns (stream UserUpdate);
}

message GetUserRequest {
    string user_id = 1;
}

message CreateUserRequest {
    string email = 1;
    string name = 2;
}

message User {
    string id = 1;
    string email = 2;
    string name = 3;
    int64 created_at = 4;
}

message ListUsersRequest {
    int32 page = 1;
    int32 page_size = 2;
}

message ListUsersResponse {
    repeated User users = 1;
    int32 total = 2;
}

message StreamRequest {
    string subscription_id = 1;
}

message UserUpdate {
    string event_type = 1;
    User user = 2;
}

gRPC Server Implementation

# pip install grpcio grpcio-tools
# python -m grpc_tools.protoc -I. --python_out=. --grpc_python_out=. proto/user_service.proto

import grpc
from concurrent import futures
import user_service_pb2
import user_service_pb2_grpc


class UserServicer(user_service_pb2_grpc.UserServiceServicer):
    """gRPC server implementation."""
    
    def __init__(self, user_repository):
        self.user_repository = user_repository
    
    async def GetUser(self, request, context):
        user = await self.user_repository.find_by_id(request.user_id)
        
        if not user:
            context.set_code(grpc.StatusCode.NOT_FOUND)
            context.set_details(f"User {request.user_id} not found")
            return user_service_pb2.User()
        
        return user_service_pb2.User(
            id=user.id,
            email=user.email,
            name=user.name,
            created_at=int(user.created_at.timestamp()),
        )
    
    async def CreateUser(self, request, context):
        try:
            user = await self.user_repository.create(
                email=request.email,
                name=request.name,
            )
            return user_service_pb2.User(
                id=user.id,
                email=user.email,
                name=user.name,
                created_at=int(user.created_at.timestamp()),
            )
        except ValueError as e:
            context.set_code(grpc.StatusCode.INVALID_ARGUMENT)
            context.set_details(str(e))
            return user_service_pb2.User()
    
    async def StreamUserUpdates(self, request, context):
        """Server streaming - push updates to client."""
        async for update in self.user_repository.watch_changes():
            yield user_service_pb2.UserUpdate(
                event_type=update.event_type,
                user=user_service_pb2.User(
                    id=update.user.id,
                    email=update.user.email,
                    name=update.user.name,
                ),
            )


async def serve():
    server = grpc.aio.server(futures.ThreadPoolExecutor(max_workers=10))
    user_service_pb2_grpc.add_UserServiceServicer_to_server(
        UserServicer(user_repository), server
    )
    server.add_insecure_port("[::]:50051")
    await server.start()
    await server.wait_for_termination()

gRPC Client

import grpc
import user_service_pb2
import user_service_pb2_grpc


class UserGRPCClient:
    """gRPC client for User Service."""
    
    def __init__(self, host: str = "user-service:50051"):
        self.channel = grpc.aio.insecure_channel(host)
        self.stub = user_service_pb2_grpc.UserServiceStub(self.channel)
    
    async def get_user(self, user_id: str) -> dict:
        try:
            response = await self.stub.GetUser(
                user_service_pb2.GetUserRequest(user_id=user_id)
            )
            return {
                "id": response.id,
                "email": response.email,
                "name": response.name,
            }
        except grpc.aio.AioRpcError as e:
            if e.code() == grpc.StatusCode.NOT_FOUND:
                raise ValueError(f"User {user_id} not found")
            raise
    
    async def watch_updates(self):
        """Stream user updates from server."""
        request = user_service_pb2.StreamRequest(subscription_id="order-service")
        async for update in self.stub.StreamUserUpdates(request):
            yield {
                "event_type": update.event_type,
                "user": {
                    "id": update.user.id,
                    "email": update.user.email,
                    "name": update.user.name,
                },
            }
    
    async def close(self):
        await self.channel.close()

REST vs gRPC Comparison

Aspect	REST/HTTP	gRPC
Protocol	HTTP/1.1, HTTP/2	HTTP/2
Payload	JSON (text)	Protocol Buffers (binary)
Performance	Good	Excellent
Streaming	Limited	Bidirectional
Browser Support	Native	Requires proxy
Tooling	Abundant	Growing
Use Case	Public APIs	Internal services

Service Discovery

Client-Side Discovery

import random
from dataclasses import dataclass
from abc import ABC, abstractmethod


@dataclass
class ServiceInstance:
    id: str
    host: str
    port: int
    metadata: dict = None
    
    @property
    def address(self) -> str:
        return f"{self.host}:{self.port}"


class ServiceRegistry(ABC):
    """Abstract service registry interface."""
    
    @abstractmethod
    async def register(self, service_name: str, instance: ServiceInstance): ...
    
    @abstractmethod
    async def deregister(self, service_name: str, instance_id: str): ...
    
    @abstractmethod
    async def get_instances(self, service_name: str) -> list[ServiceInstance]: ...


class InMemoryServiceRegistry(ServiceRegistry):
    """Simple in-memory registry for development."""
    
    def __init__(self):
        self.services: dict[str, dict[str, ServiceInstance]] = {}
    
    async def register(self, service_name: str, instance: ServiceInstance):
        if service_name not in self.services:
            self.services[service_name] = {}
        self.services[service_name][instance.id] = instance
    
    async def deregister(self, service_name: str, instance_id: str):
        if service_name in self.services:
            self.services[service_name].pop(instance_id, None)
    
    async def get_instances(self, service_name: str) -> list[ServiceInstance]:
        return list(self.services.get(service_name, {}).values())


class DiscoveryClient:
    """Client that uses service discovery for routing."""
    
    def __init__(self, registry: ServiceRegistry):
        self.registry = registry
        self.http_client = httpx.AsyncClient()
    
    async def call(
        self, 
        service_name: str, 
        method: str, 
        path: str, 
        **kwargs
    ) -> httpx.Response:
        instances = await self.registry.get_instances(service_name)
        
        if not instances:
            raise ServiceError(
                error_type=ServiceErrorType.SERVICE_UNAVAILABLE,
                message=f"No instances available for {service_name}",
                service=service_name,
            )
        
        # Simple random load balancing
        instance = random.choice(instances)
        url = f"http://{instance.address}{path}"
        
        return await self.http_client.request(method, url, **kwargs)

DNS-Based Discovery

import socket
import asyncio


class DNSServiceDiscovery:
    """DNS-based service discovery (works with Kubernetes, Consul DNS)."""
    
    def __init__(self, dns_suffix: str = "svc.cluster.local"):
        self.dns_suffix = dns_suffix
    
    async def resolve(self, service_name: str) -> list[str]:
        """Resolve service name to IP addresses."""
        hostname = f"{service_name}.{self.dns_suffix}"
        
        loop = asyncio.get_event_loop()
        try:
            # DNS SRV record lookup for port info
            # Or simple A record lookup
            result = await loop.getaddrinfo(
                hostname, None, 
                family=socket.AF_INET,
                type=socket.SOCK_STREAM,
            )
            return [addr[4][0] for addr in result]
        except socket.gaierror:
            return []


class DNSAwareClient:
    """HTTP client with DNS-based discovery."""
    
    def __init__(self, discovery: DNSServiceDiscovery):
        self.discovery = discovery
        self.client = httpx.AsyncClient()
    
    async def get(self, service: str, path: str) -> dict:
        addresses = await self.discovery.resolve(service)
        if not addresses:
            raise ServiceError(
                error_type=ServiceErrorType.SERVICE_UNAVAILABLE,
                message=f"Cannot resolve {service}",
                service=service,
            )
        
        # Round-robin or random selection
        address = random.choice(addresses)
        response = await self.client.get(f"http://{address}:8000{path}")
        response.raise_for_status()
        return response.json()

Load Balancing

Client-Side Load Balancing

from abc import ABC, abstractmethod
from collections import deque
import time


class LoadBalancer(ABC):
    """Abstract load balancer interface."""
    
    @abstractmethod
    def select(self, instances: list[ServiceInstance]) -> ServiceInstance: ...


class RoundRobinLoadBalancer(LoadBalancer):
    """Round-robin load balancing."""
    
    def __init__(self):
        self.index = 0
    
    def select(self, instances: list[ServiceInstance]) -> ServiceInstance:
        if not instances:
            raise ValueError("No instances available")
        
        instance = instances[self.index % len(instances)]
        self.index += 1
        return instance


class WeightedLoadBalancer(LoadBalancer):
    """Weighted load balancing based on instance capacity."""
    
    def select(self, instances: list[ServiceInstance]) -> ServiceInstance:
        if not instances:
            raise ValueError("No instances available")
        
        # Use metadata for weights
        weighted_instances = []
        for instance in instances:
            weight = instance.metadata.get("weight", 1) if instance.metadata else 1
            weighted_instances.extend([instance] * weight)
        
        return random.choice(weighted_instances)


class LeastConnectionsLoadBalancer(LoadBalancer):
    """Route to instance with fewest active connections."""
    
    def __init__(self):
        self.connections: dict[str, int] = {}
    
    def select(self, instances: list[ServiceInstance]) -> ServiceInstance:
        if not instances:
            raise ValueError("No instances available")
        
        # Find instance with minimum connections
        min_instance = min(
            instances,
            key=lambda i: self.connections.get(i.id, 0)
        )
        return min_instance
    
    def increment(self, instance_id: str):
        self.connections[instance_id] = self.connections.get(instance_id, 0) + 1
    
    def decrement(self, instance_id: str):
        self.connections[instance_id] = max(0, self.connections.get(instance_id, 0) - 1)


class LoadBalancedClient:
    """HTTP client with load balancing."""
    
    def __init__(
        self, 
        registry: ServiceRegistry,
        load_balancer: LoadBalancer | None = None,
    ):
        self.registry = registry
        self.load_balancer = load_balancer or RoundRobinLoadBalancer()
        self.client = httpx.AsyncClient()
    
    async def request(
        self, 
        service_name: str, 
        method: str, 
        path: str, 
        **kwargs
    ) -> httpx.Response:
        instances = await self.registry.get_instances(service_name)
        instance = self.load_balancer.select(instances)
        
        url = f"http://{instance.address}{path}"
        return await self.client.request(method, url, **kwargs)

Health-Aware Load Balancing

from dataclasses import dataclass, field
from datetime import datetime, timedelta


@dataclass
class InstanceHealth:
    instance_id: str
    healthy: bool = True
    last_check: datetime = field(default_factory=datetime.utcnow)
    consecutive_failures: int = 0
    
    def mark_healthy(self):
        self.healthy = True
        self.consecutive_failures = 0
        self.last_check = datetime.utcnow()
    
    def mark_unhealthy(self):
        self.consecutive_failures += 1
        self.last_check = datetime.utcnow()
        if self.consecutive_failures >= 3:
            self.healthy = False


class HealthAwareLoadBalancer(LoadBalancer):
    """Load balancer that considers instance health."""
    
    def __init__(self, check_interval: timedelta = timedelta(seconds=30)):
        self.health_status: dict[str, InstanceHealth] = {}
        self.check_interval = check_interval
        self.inner_lb = RoundRobinLoadBalancer()
    
    def select(self, instances: list[ServiceInstance]) -> ServiceInstance:
        # Filter to healthy instances
        healthy_instances = [
            i for i in instances
            if self._is_healthy(i.id)
        ]
        
        if not healthy_instances:
            # If no healthy instances, try all (maybe health checks are stale)
            healthy_instances = instances
        
        return self.inner_lb.select(healthy_instances)
    
    def _is_healthy(self, instance_id: str) -> bool:
        health = self.health_status.get(instance_id)
        if not health:
            return True  # Assume healthy if unknown
        return health.healthy
    
    def report_success(self, instance_id: str):
        if instance_id not in self.health_status:
            self.health_status[instance_id] = InstanceHealth(instance_id)
        self.health_status[instance_id].mark_healthy()
    
    def report_failure(self, instance_id: str):
        if instance_id not in self.health_status:
            self.health_status[instance_id] = InstanceHealth(instance_id)
        self.health_status[instance_id].mark_unhealthy()

Practical Patterns

Timeout and Deadline Propagation

import time
from contextvars import ContextVar

deadline_var: ContextVar[float | None] = ContextVar("deadline", default=None)


def set_deadline(timeout_seconds: float):
    """Set request deadline."""
    deadline_var.set(time.time() + timeout_seconds)


def remaining_time() -> float | None:
    """Get remaining time until deadline."""
    deadline = deadline_var.get()
    if deadline is None:
        return None
    remaining = deadline - time.time()
    return max(0, remaining)


class DeadlineAwareClient:
    """Client that propagates deadlines across services."""
    
    async def call(self, service: str, path: str) -> dict:
        remaining = remaining_time()
        
        if remaining is not None and remaining <= 0:
            raise TimeoutError("Request deadline exceeded")
        
        timeout = min(remaining, 30.0) if remaining else 30.0
        
        async with httpx.AsyncClient(timeout=timeout) as client:
            response = await client.get(f"http://{service}{path}")
            return response.json()


# Usage in request handler
@app.get("/orders/{order_id}")
async def get_order(order_id: str):
    set_deadline(5.0)  # 5 second deadline
    
    # Each downstream call respects the deadline
    user = await user_client.call("user-service", f"/users/{order.user_id}")
    products = await product_client.call("product-service", "/products/batch")
    
    return {"order": order, "user": user, "products": products}

Request Hedging

import asyncio


class HedgedClient:
    """Client that hedges requests for lower latency."""
    
    def __init__(self, registry: ServiceRegistry, hedge_delay: float = 0.1):
        self.registry = registry
        self.hedge_delay = hedge_delay
        self.client = httpx.AsyncClient()
    
    async def get_with_hedging(self, service: str, path: str) -> dict:
        """Send hedged requests - first response wins."""
        instances = await self.registry.get_instances(service)
        
        if len(instances) < 2:
            # Not enough instances to hedge
            instance = instances[0]
            response = await self.client.get(f"http://{instance.address}{path}")
            return response.json()
        
        # Create tasks for multiple instances
        async def make_request(instance: ServiceInstance):
            response = await self.client.get(f"http://{instance.address}{path}")
            return response.json()
        
        # Start first request immediately
        tasks = [asyncio.create_task(make_request(instances[0]))]
        
        # Start hedge request after delay
        await asyncio.sleep(self.hedge_delay)
        tasks.append(asyncio.create_task(make_request(instances[1])))
        
        # Return first completed result
        done, pending = await asyncio.wait(
            tasks,
            return_when=asyncio.FIRST_COMPLETED,
        )
        
        # Cancel pending tasks
        for task in pending:
            task.cancel()
        
        return done.pop().result()

Practical Exercise

Exercise: Build a Resilient Service Client

from dataclasses import dataclass
from typing import Callable, Any
import httpx
import asyncio


@dataclass
class ClientConfig:
    base_url: str
    timeout: float = 30.0
    max_retries: int = 3
    retry_delay: float = 1.0


class ResilientClient:
    """Production-ready service client with all patterns."""
    
    def __init__(self, config: ClientConfig, service_name: str):
        self.config = config
        self.service_name = service_name
        self.client = httpx.AsyncClient(
            base_url=config.base_url,
            timeout=httpx.Timeout(config.timeout),
        )
    
    async def get(self, path: str, **kwargs) -> dict:
        return await self._request_with_retry("GET", path, **kwargs)
    
    async def post(self, path: str, data: dict, **kwargs) -> dict:
        return await self._request_with_retry("POST", path, json=data, **kwargs)
    
    async def _request_with_retry(
        self, 
        method: str, 
        path: str, 
        **kwargs
    ) -> dict:
        last_error = None
        
        for attempt in range(self.config.max_retries):
            try:
                response = await self.client.request(method, path, **kwargs)
                response.raise_for_status()
                return response.json()
            
            except httpx.HTTPStatusError as e:
                if e.response.status_code < 500:
                    # Don't retry client errors
                    raise
                last_error = e
            
            except (httpx.ConnectError, httpx.TimeoutException) as e:
                last_error = e
            
            # Exponential backoff
            if attempt < self.config.max_retries - 1:
                delay = self.config.retry_delay * (2 ** attempt)
                await asyncio.sleep(delay)
        
        raise last_error
    
    async def close(self):
        await self.client.aclose()


# Usage
async def main():
    client = ResilientClient(
        ClientConfig(base_url="http://user-service:8000"),
        service_name="user-service",
    )
    
    try:
        user = await client.get("/users/123")
        print(f"User: {user}")
    finally:
        await client.close()

Key Takeaways

1. HTTP/REST for simplicity - Use for most service-to-service communication

2. gRPC for performance - Consider for high-throughput internal APIs

3. Propagate context - Trace IDs, deadlines, and correlation IDs

4. Handle failures gracefully - Proper error classification and handling

5. Use load balancing - Distribute traffic across instances

What's Next?

Synchronous communication creates coupling and cascading failures. In Article 5: Asynchronous Communication, we'll explore message queues, event-driven patterns, and eventual consistency.

---

This article is part of the Microservice Architecture 101 series.