gRPC vs REST — Unbounded Stream Buffers Cause OOM

// ── 1. Define the contract in .proto file ──────────────────────────────────
// File: src/main/proto/payment.proto
/*
syntax = "proto3";
option java_package = "io.thecodeforge.payment.grpc";
option java_outer_classname = "PaymentProto";

service PaymentService {
  // Unary RPC — one request, one response (like REST)
  rpc ProcessPayment(PaymentRequest) returns (PaymentResponse);

  // Server streaming — one request, stream of responses
  rpc StreamTransactions(TransactionFilter) returns (stream Transaction);

  // Client streaming — stream of requests, one response
  rpc BatchPayments(stream PaymentRequest) returns (BatchResult);

  // Bidirectional streaming — both sides stream
  rpc PaymentFeed(stream PaymentEvent) returns (stream PaymentEvent);
}

message PaymentRequest {
  string customer_id = 1;
  int64 amount_pence = 2;
  string currency    = 3;
}

message PaymentResponse {
  string payment_id = 1;
  string status     = 2;
  int64  timestamp  = 3;
}
*/

// ── 2. Generated server implementation ─────────────────────────────────────
package io.thecodeforge.payment.grpc;

import io.grpc.stub.StreamObserver;

public class PaymentServiceImpl
    extends PaymentServiceGrpc.PaymentServiceImplBase {

    @Override
    public void processPayment(
            PaymentRequest request,
            StreamObserver<PaymentResponse> responseObserver) {

        // Business logic
        String paymentId = processInternally(
            request.getCustomerId(),
            request.getAmountPence(),
            request.getCurrency()
        );

        // Build protobuf response (binary, not JSON)
        PaymentResponse response = PaymentResponse.newBuilder()
            .setPaymentId(paymentId)
            .setStatus("COMPLETED")
            .setTimestamp(System.currentTimeMillis())
            .build();

        responseObserver.onNext(response);
        responseObserver.onCompleted();
    }

    private String processInternally(String customerId, long amount, String currency) {
        // Payment processing logic
        return "pay-" + System.nanoTime();
    }
}

// ── 3. Client usage — generated stub handles serialisation ─────────────────
package io.thecodeforge.payment.grpc;

import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;

public class PaymentClient {
    public static void main(String[] args) {
        ManagedChannel channel = ManagedChannelBuilder
            .forAddress("payment-service.internal", 50051)
            .usePlaintext()
            .build();

        PaymentServiceGrpc.PaymentServiceBlockingStub stub =
            PaymentServiceGrpc.newBlockingStub(channel);

        PaymentRequest request = PaymentRequest.newBuilder()
            .setCustomerId("customer-42")
            .setAmountPence(10000)  // £100.00
            .setCurrency("GBP")
            .build();

        PaymentResponse response = stub.processPayment(request);
        System.out.println("Payment ID: " + response.getPaymentId());
        System.out.println("Status: "    + response.getStatus());
    }
}

package io.thecodeforge.payment.rest;

import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;

@RestController
@RequestMapping("/api/v1/payments")
public class PaymentRestController {

    private final PaymentService paymentService;

    public PaymentRestController(PaymentService paymentService) {
        this.paymentService = paymentService;
    }

    @PostMapping
    public ResponseEntity<PaymentResponse> processPayment(
            @RequestBody PaymentRequest request) {

        // REST: consumed by any HTTP client, browser, curl, Postman
        // No code generation required. Request/response in JSON.
        PaymentResult result = paymentService.process(
            request.getCustomerId(),
            request.getAmountPence(),
            request.getCurrency()
        );

        return ResponseEntity.ok(new PaymentResponse(
            result.getPaymentId(),
            result.getStatus().name(),
            result.getTimestamp()
        ));
    }

    @GetMapping("/{paymentId}")
    public ResponseEntity<PaymentResponse> getPayment(@PathVariable String paymentId) {
        return paymentService.findById(paymentId)
            .map(p -> ResponseEntity.ok(new PaymentResponse(p)))
            .orElse(ResponseEntity.notFound().build());
    }
}

// curl -X POST https://api.thecodeforge.io/v1/payments \
//   -H 'Content-Type: application/json' \
//   -d '{"customerId":"c-42","amountPence":10000,"currency":"GBP"}'
//
// Response:
// {"paymentId":"pay-1234","status":"COMPLETED","timestamp":1711756800000}

package io.thecodeforge.payment.grpc;

import io.grpc.stub.StreamObserver;
import java.util.List;

public class TransactionStreamingServiceImpl
    extends TransactionServiceGrpc.TransactionServiceImplBase {

    private final TransactionRepository transactionRepo;

    public TransactionStreamingServiceImpl(TransactionRepository repo) {
        this.transactionRepo = repo;
    }

    // Server streaming: client sends one filter, server streams matching transactions
    // Useful for: exporting large datasets without pagination loops
    @Override
    public void streamTransactions(
            TransactionFilter filter,
            StreamObserver<Transaction> responseObserver) {

        // Fetch and stream — client receives each transaction as it's sent
        // No need to buffer the entire result set in memory
        transactionRepo.findByCustomerIdAndDateRange(
            filter.getCustomerId(),
            filter.getFromTimestamp(),
            filter.getToTimestamp()
        ).forEach(tx -> {
            Transaction proto = Transaction.newBuilder()
                .setTransactionId(tx.getId())
                .setAmountPence(tx.getAmountPence())
                .setCurrency(tx.getCurrency())
                .setTimestamp(tx.getTimestamp())
                .build();
            responseObserver.onNext(proto);  // Each transaction sent immediately
        });

        responseObserver.onCompleted();
    }

    // Client streaming: client sends batch of payments, server responds once
    @Override
    public StreamObserver<PaymentRequest> batchPayments(
            StreamObserver<BatchResult> responseObserver) {

        return new StreamObserver<>() {
            int processed = 0;
            int failed = 0;

            @Override
            public void onNext(PaymentRequest request) {
                // Process each payment as it arrives from client
                try {
                    processPayment(request);
                    processed++;
                } catch (Exception e) {
                    failed++;
                }
            }

            @Override
            public void onCompleted() {
                // Client done sending — send summary response
                responseObserver.onNext(BatchResult.newBuilder()
                    .setProcessed(processed)
                    .setFailed(failed)
                    .build());
                responseObserver.onCompleted();
            }

            @Override
            public void onError(Throwable t) {
                // Handle client-side error
            }
        };
    }
}

package io.thecodeforge.payment.grpc;

import io.grpc.Status;
import io.grpc.StatusRuntimeException;
import io.grpc.protobuf.StatusProto;
import com.google.rpc.Code;
import com.google.rpc.RetryInfo;
import com.google.protobuf.Duration;

public class GrpcErrorHandling {

    // ❌ Wrong: throwing generic Status.INTERNAL
    public void processPaymentBad(String id) {
        if (id == null) {
            throw Status.INTERNAL
                .withDescription("null customer id")
                .asRuntimeException();
        }
        // This gives no indication to client that it's a client error
    }

    // ✅ Correct: using the specific status code with rich error details
    public void processPaymentGood(String id) {
        if (id == null) {
            // Build a structured error with retry information
            com.google.rpc.Status status = com.google.rpc.Status.newBuilder()
                .setCode(Code.INVALID_ARGUMENT.getNumber())
                .setMessage("customer id must not be null")
                .addDetails(
                    RetryInfo.newBuilder()
                        .setRetryDelay(Duration.newBuilder()
                            .setSeconds(5)
                            .build())
                        .build()
                )
                .build();
            throw StatusProto.toStatusRuntimeException(status);
        }
    }

    // Client side: different codes lead to different recovery
    public static void main(String[] args) {
        try {
            // call grpc service
        } catch (StatusRuntimeException e) {
            switch (e.getStatus().getCode()) {
                case INVALID_ARGUMENT:
                    // Log and stop retrying — bad request
                    break;
                case UNAVAILABLE:
                    // Retry with backoff — transient
                    break;
                case DEADLINE_EXCEEDED:
                    // Increase timeout or check downstream
                    break;
            }
        }
    }
}

#!/bin/bash
# Debugging tools comparison

# REST: universally available
curl https://api.thecodeforge.io/v1/payments/pay-123 -H 'Accept: application/json'

# gRPC: requires grpcurl and reflection or proto descriptors
grpcurl -plaintext payment-service.internal:50051 list
grpcurl -plaintext payment-service.internal:50051 io.thecodeforge.payment.grpc.PaymentService/ProcessPayment

# REST: easy to inspect in browser devtools
echo "Open Chrome DevTools → Network tab → any request"

# gRPC: need proxy or packet capture
echo "Use tcpdump or grpcurl with -d '{}' to send test calls"

# REST: metrics in every monitoring tool
curl http://prometheus:9090/api/v1/query?query=http_requests_total{handler="/api/v1/payments"}

# gRPC: metrics require custom interceptors (micrometer or prometheus client)
echo "Add gRPC server interceptor: new MetricsServerInterceptor(...)"

// io.thecodeforge — system-design tutorial

import grpc
from grpc_interceptor import ServerInterceptor
import jwt

class JwtValidationInterceptor(ServerInterceptor):
    def intercept(self, method, request, context, method_name):
        # Skip auth for health check — every gRPC service has one
        if method_name == "/grpc.health.v1.Health/Check":
            return method(request, context)

        metadata = dict(context.invocation_metadata())
        token = metadata.get("authorization", "").replace("Bearer ", "")

        if not token:
            context.abort(grpc.StatusCode.UNAUTHENTICATED, "Missing token")

        try:
            decoded = jwt.decode(token, "secret", algorithms=["HS256"])
            return method(request, context)
        except jwt.ExpiredSignatureError:
            context.abort(grpc.StatusCode.UNAUTHENTICATED, "Token expired")
        except jwt.InvalidTokenError:
            context.abort(grpc.StatusCode.UNAUTHENTICATED, "Invalid token")

// io.thecodeforge — system-design tutorial

import grpc

# Client-side load balancing with DNS and round_robin strategy
# Requires headless service in Kubernetes
channel = grpc.insecure_channel(
    target="dns:///my-grpc-service.default.svc.cluster.local:50051",
    options=[
        ("grpc.lb_policy_name", "round_robin"),  # Spread streams across backends
        ("grpc.dns_min_time_between_resolutions_ms", 10000),  # Re-resolve every 10s
    ]
)

creds = grpc.ssl_channel_credentials()
secured_channel = grpc.secure_channel(
    target="dns:///my-grpc-service.default.svc.cluster.local:50051",
    credentials=creds,
    options=[
        ("grpc.lb_policy_name", "pick_first"),  # Default — dangerous for perf
    ]
)

// io.thecodeforge — system-design tutorial

import grpc
from grpc_interceptor import ServerInterceptor

class TokenLeakCatcher(ServerInterceptor):
    def intercept(self, method, request, context, method_name):
        metadata = dict(context.invocation_metadata())
        if "authorization" in metadata:
            token = metadata["authorization"]
            if token.startswith("Bearer "):
                # In production: redact before logging
                context.set_trailing_metadata(("x-token-hash", hash(token)))
        return method(request, context, method_name)

// io.thecodeforge — system-design tutorial

import grpc
import market_data_pb2_grpc as pb2_grpc
import market_data_pb2 as pb2

channel = grpc.insecure_channel("ticker-service:50051")
stub = pb2_grpc.MarketDataStub(channel)

for tick in stub.StreamTrades(
    pb2.TradeFilter(symbol="AAPL", exchange="NASDAQ")
):
    print(f"{tick.price} @ {tick.volume} ({tick.timestamp})")

// io.thecodeforge — system-design tutorial

import json
from google.protobuf import json_format

# REST: optional fields survive silently
rest_payload = '{"name": "Alice"}'  # no age field
# gRPC: missing required field -> error
from person_pb2 import Person
p = Person()
p.name = "Alice"
p.age = 30  # if required in proto, must be set

// io.thecodeforge — system-design tutorial

# REST convention: 2xx success, 4xx client error
# But body format is team-defined
response = {"status": "ok", "data": {}}

# gRPC rule: every RPC returns typed response or error
import grpc
from google.rpc import code_pb2
# No room for interpretation — status is numeric enum