Sync vs Async Communication — REST, gRPC & Message Queues

Synchronous (Request-Response):
  Service A ──── request ────→ Service B
  Service A ←─── response ─── Service B
  Service A blocks until B responds

Asynchronous (Fire and Forget):
  Service A ──── message ────→ Message Broker
  Service A continues (doesn't wait)
                               Message Broker ──→ Service B (eventually)

GET    /orders/{id}           → get order by ID
POST   /orders                → create order
PUT    /orders/{id}/status    → update order status
DELETE /orders/{id}           → cancel order

Versioning:
  /api/v1/orders  → version in URL (simple, visible)
  Accept: application/vnd.myapi.v1+json → version in header (clean URLs)

JSON over HTTP/1.1:
  Request:  ~200 bytes (text, headers)
  Parse:    ~0.5ms (JSON deserialization)

Protobuf over HTTP/2:
  Request:  ~50 bytes (binary, compressed)
  Parse:    ~0.05ms (binary deserialization)
  
10× smaller messages, 10× faster parsing

// order.proto
syntax = "proto3";

service OrderService {
  rpc GetOrder (GetOrderRequest) returns (Order);
  rpc ListOrders (ListOrdersRequest) returns (stream Order);  // server streaming
  rpc CreateOrder (CreateOrderRequest) returns (CreateOrderResponse);
}

message Order {
  string id = 1;
  string user_id = 2;
  OrderStatus status = 3;
  repeated OrderItem items = 4;
  google.protobuf.Timestamp created_at = 5;
}

enum OrderStatus {
  PENDING = 0;
  CONFIRMED = 1;
  SHIPPED = 2;
  DELIVERED = 3;
}

// .NET gRPC service implementation
public class OrderGrpcService : OrderService.OrderServiceBase
{
    private readonly IOrderRepository _orders;
    
    public override async Task<Order> GetOrder(
        GetOrderRequest request, ServerCallContext context)
    {
        var order = await _orders.GetByIdAsync(request.Id);
        if (order == null)
            throw new RpcException(new Status(StatusCode.NotFound, "Order not found"));
        
        return MapToProto(order);
    }
    
    // Server-side streaming: push multiple responses for one request
    public override async Task ListOrders(
        ListOrdersRequest request,
        IServerStreamWriter<Order> responseStream,
        ServerCallContext context)
    {
        await foreach (var order in _orders.StreamByUserAsync(request.UserId))
        {
            await responseStream.WriteAsync(MapToProto(order));
        }
    }
}

Topology:
  Publisher → Exchange → (routing rule) → Queue → Consumer

Exchange types:
  Direct:  exact routing key match
  Topic:   routing key pattern (order.created, order.*)
  Fanout:  broadcast to all bound queues
  Headers: route by message headers

// Publishing to RabbitMQ
public class OrderEventPublisher
{
    private readonly IModel _channel;
    
    public void PublishOrderCreated(OrderCreatedEvent @event)
    {
        var body = JsonSerializer.SerializeToUtf8Bytes(@event);
        
        _channel.BasicPublish(
            exchange: "order-events",
            routingKey: "order.created",
            basicProperties: null,
            body: body
        );
    }
}

// Consuming from RabbitMQ
public class NotificationConsumer : BackgroundService
{
    protected override Task ExecuteAsync(CancellationToken ct)
    {
        _channel.QueueDeclare("notification-queue", durable: true);
        _channel.QueueBind("notification-queue", "order-events", "order.created");
        
        var consumer = new EventingBasicConsumer(_channel);
        consumer.Received += async (_, ea) =>
        {
            var @event = JsonSerializer.Deserialize<OrderCreatedEvent>(ea.Body.ToArray());
            await _notificationService.SendOrderConfirmationAsync(@event);
            _channel.BasicAck(ea.DeliveryTag, multiple: false);  // acknowledge
        };
        
        _channel.BasicConsume("notification-queue", autoAck: false, consumer);
        return Task.CompletedTask;
    }
}

// Azure Service Bus sender
var client = new ServiceBusClient(connectionString);
var sender = client.CreateSender("order-events");

var message = new ServiceBusMessage(
    JsonSerializer.SerializeToUtf8Bytes(orderCreatedEvent))
{
    ContentType = "application/json",
    CorrelationId = correlationId,
    SessionId = order.UserId,  // session-based ordering per user
};

await sender.SendMessageAsync(message);

Kafka key concepts:
  Topic: a named log (e.g., "order-events")
  Partition: topics split into partitions for parallelism
  Offset: position of a message in a partition
  Consumer Group: multiple consumers sharing topic consumption
  
Partitioning by key:
  Kafka routes messages with the same key to the same partition
  This guarantees ordering per key (e.g., per user_id or order_id)
  
                  Partition 0: user_1 messages
  Topic ─────────→ Partition 1: user_2 messages
                  Partition 2: user_3 messages

Message: "ProcessPayment" for order #123 → Payment Service
Semantics: "Please do this"
If Payment Service is down, this is a problem

Message: "OrderPlaced" for order #123
Semantics: "This happened. Do what you need."
Publisher doesn't know or care who reacts
Notification Service, Inventory Service, Analytics Service all subscribe

OrderPlaced event published
  → Inventory Service reacts: reserve stock → InventoryReserved event
  → InventoryReserved event published
  → Payment Service reacts: charge card → PaymentProcessed event
  → PaymentProcessed event published
  → Notification Service reacts: send email
  
No one is "in charge". Services are decoupled.

// Order Saga Orchestrator
public class CreateOrderSaga
{
    public async Task ExecuteAsync(CreateOrderCommand command)
    {
        // Step 1: Reserve inventory
        var reservation = await _inventoryService.ReserveAsync(command.Items);
        
        if (!reservation.Success)
        {
            await _orderService.MarkFailedAsync(command.OrderId, "Inventory unavailable");
            return;
        }
        
        // Step 2: Process payment
        var payment = await _paymentService.ChargeAsync(command.UserId, command.Total);
        
        if (!payment.Success)
        {
            // Compensate: release inventory
            await _inventoryService.ReleaseAsync(reservation.Id);
            await _orderService.MarkFailedAsync(command.OrderId, "Payment failed");
            return;
        }
        
        // Step 3: Confirm order
        await _orderService.ConfirmAsync(command.OrderId, payment.TransactionId);
        
        // Step 4: Notify (fire and forget — not business-critical)
        await _notificationService.SendConfirmationAsync(command.UserId, command.OrderId);
    }
}

The problem:
  1. Update order status in DB  ✓
  2. Publish "OrderShipped" event to Kafka  ✗ (Kafka is down)
  
  → DB updated, event never published
  → Notification never sent, inventory never updated
  → Data inconsistency

1. In the SAME database transaction:
   - Update order status
   - INSERT into outbox table: {event: "OrderShipped", payload: {...}, published: false}
   
2. Separate background worker (Outbox Processor):
   - Polls outbox table for unpublished events
   - Publishes each event to Kafka/Service Bus
   - Marks event as published in outbox table
   
Result: Either both the order update and the outbox entry commit,
        or neither does. The outbox processor handles eventual publishing.

// Outbox table
CREATE TABLE outbox_events (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    event_type VARCHAR(100) NOT NULL,
    payload JSONB NOT NULL,
    created_at TIMESTAMP DEFAULT NOW(),
    published_at TIMESTAMP NULL,
    retry_count INT DEFAULT 0
);

// Service layer — atomic: DB update + outbox insert
public async Task ShipOrderAsync(Guid orderId)
{
    using var tx = await _db.BeginTransactionAsync();
    
    await _db.ExecuteAsync(
        "UPDATE orders SET status = 'Shipped', shipped_at = NOW() WHERE id = @id",
        new { id = orderId });
    
    await _db.ExecuteAsync(
        "INSERT INTO outbox_events (event_type, payload) VALUES (@type, @payload::jsonb)",
        new { type = "OrderShipped", payload = JsonSerializer.Serialize(new { OrderId = orderId }) });
    
    await tx.CommitAsync();
    // Even if Kafka is down, the event is safely stored in the DB
}

Incoming request:
  POST /checkout  →  X-Correlation-ID: req-abc123

Service A logs:  [req-abc123] Created order
Service B logs:  [req-abc123] Reserved inventory
Service C logs:  [req-abc123] Charged payment
Service D logs:  [req-abc123] Sent confirmation email

Without correlation IDs:
  When checkout fails, you search logs across 4 services with no link.
  With correlation IDs: grep for "req-abc123" and see the full trace.

// ASP.NET Core middleware — propagate correlation ID
public class CorrelationIdMiddleware
{
    public async Task InvokeAsync(HttpContext context)
    {
        var correlationId = context.Request.Headers["X-Correlation-ID"].FirstOrDefault()
            ?? Guid.NewGuid().ToString();
        
        context.Items["CorrelationId"] = correlationId;
        context.Response.Headers["X-Correlation-ID"] = correlationId;
        
        // Add to outgoing HTTP client requests
        using var scope = _logger.BeginScope(new { CorrelationId = correlationId });
        await _next(context);
    }
}

// HttpClient factory — inject correlation ID into downstream calls
services.AddHttpClient("order-service")
    .AddHttpMessageHandler<CorrelationIdDelegatingHandler>();

Boundary                           Protocol
────────────────────────────────────────────────────────────
API Gateway → Services             REST (public-facing, tooling)
Service → Service (sync needed)    gRPC (internal, typed, fast)
Service → Service (async ok)       Message Queue (decoupled, resilient)
Order placed → Notification        Event (async, publish-subscribe)
Checkout → Inventory               gRPC or REST (sync, need immediate confirmation)
Order shipped → Notify/Analytics   Event via Kafka (fanout, replay needed)