---

name: grpc-patterns description: Implements gRPC service patterns (unary, streaming, bidirectional), Protocol Buffers design, interceptor middleware, typed error handling, and client/server code generation for Go and Python microservices. license: MIT compatibility: opencode metadata: version: "1.0.0" domain: coding triggers: gRPC, protocol buffers, protobuf, RPC, streaming RPC, unary call, interceptor middleware, grpc service middleware archetypes:

• tactical
• generation anti_triggers:
• brainstorming
• vague ideation
• code golf
• over-engineering response_profile: verbosity: low directive_strength: high abstraction_level: operational role: implementation scope: implementation output-format: code content-types:
• code
• guidance
• do-dont
• examples related-skills: grpc, rest-api-patterns, fastapi-patterns

---

gRPC Service Patterns

Implements production-ready gRPC services and clients using Protocol Buffers, covering unary and streaming RPCs, interceptor middleware, typed error handling with status codes, and cross-language code generation for Go and Python.

TL;DR Checklist

• Define .proto service contract first — write methods, messages, and enum types before any implementation
• Use versioned package names (package myapp.v1) and never reuse field numbers across versions
• Map all error paths to specific gRPC status codes (InvalidArgument, NotFound, AlreadyExists, etc.)
• Set context deadlines on every client call — no calls without timeouts
• Implement interceptor chains for cross-cutting concerns (auth, logging, metrics)
• Handle stream lifecycle explicitly — always check for EOF and cancel contexts

---

When to Use

Use this skill when:

• Writing gRPC service implementations in Go or Python that need production-grade patterns
• Designing .proto schemas for new microservices or evolving existing ones
• Implementing interceptor middleware chains (authentication, logging, retry logic)
• Building streaming RPC endpoints (server-streaming, client-streaming, bidirectional)
• Setting up gRPC clients with proper error handling, timeouts, and interceptors
• Migrating a REST API to gRPC or designing internal service-to-service communication

---

When NOT to Use

Avoid this skill for:

• External public API design — use REST/GraphQL for broader client compatibility (see rest-api-patterns)
• Kubernetes deployment configuration — use cncf/grpc for manifests, ServiceMesh routing, and Envoy configs
• Simple file format serialization without RPC needs — use JSON or Protocol Buffers directly without gRPC transport

---

Core Workflow

1. Design the .proto Schema — Define service methods, message types, enums, and options in .proto files first. Use package myapp.v1 naming for versioning. Assign stable field numbers and never reuse them. Include google.api.http annotations if you need HTTP/gRPC gateway later. Checkpoint: Every RPC method has a clearly named request and response message — no bare string parameters, no untyped maps for structured data.

2. Generate Service Stubs — Run protoc with language-specific plugins to generate server stubs and client libraries. For Go: protoc --go_out=. --go-grpc_out=.. For Python: python -m grpc_tools.protoc. Commit generated files or ensure CI generates them consistently. Checkpoint: Generated code compiles cleanly in the target language and matches the proto definition.

3. Implement Service Handlers — Implement each RPC method with input validation, business logic, and proper gRPC status code returns. Never return raw errors as strings — wrap them using status.Error() (Go) or grpc.StatusCode (Python). Checkpoint: Every error path maps to a specific gRPC status code, not a generic Internal error.

4. Build Interceptor Chains — Implement server-side interceptors for authentication, request validation, and metrics. Implement client-side interceptors for retry logic and logging. Chain them in order: auth → validate → timeout → call → log metrics. Checkpoint: Each interceptor is independent and testable in isolation.

5. Implement Client Code — Create gRPC clients with context deadlines, configurable interceptors, connection pool settings, and proper error handling. Use exponential backoff retry for transient errors (Unavailable, DeadlineExceeded). Checkpoint: All client calls have a deadline or timeout — no unbounded waits.

---

Implementation Patterns

Pattern 1: Protobuf Schema Design

Use versioned packages, stable field numbers, and clear naming conventions. Prefer int64 over int32 for IDs to avoid overflow in long-lived systems.

// ✅ GOOD — versioned package with clear types
syntax = "proto3";

package orders.v1;

option go_package = "github.com/myorg/orders/proto/gen/orders/v1;ordersv1";
option java_package = "com.myorg.orders.v1";

import "google/protobuf/timestamp.proto";
import "google/api/annotations.proto";

// Enum values should start at 0 and use UPPER_SNAKE_CASE
enum OrderStatus {
  ORDER_STATUS_UNSPECIFIED = 0;
  ORDER_STATUS_PENDING = 1;
  ORDER_STATUS_CONFIRMED = 2;
  ORDER_STATUS_SHIPPED = 3;
  ORDER_STATUS_DELIVERED = 4;
  ORDER_STATUS_CANCELLED = 5;
}

// Request/response messages named after the RPC method + action
message CreateOrderRequest {
  string user_id = 1;
  repeated OrderItem items = 2;
  string shipping_address_id = 3;
}

message OrderItem {
  string product_id = 1;
  int32 quantity = 2;
  int64 unit_price_cents = 3;  // int64 for precision, avoids float
}

message CreateOrderResponse {
  string order_id = 1;
  google.protobuf.Timestamp created_at = 2;
  OrderStatus status = 3;
}

// Service with HTTP gateway annotations for REST-to-gRPC bridge
service OrderService {
  rpc CreateOrder(CreateOrderRequest) returns (CreateOrderResponse) {
    option (google.api.http) = {
      post: "/v1/orders"
      body: "*"
    };
  }

  rpc GetOrder(GetOrderRequest) returns (Order) {
    option (google.api.http) = {
      get: "/v1/orders/{order_id}"
    };
  }

  // Server streaming — tracks order status updates in real time
  rpc TrackOrder(TrackOrderRequest) returns (stream OrderStatusUpdate);

  // Client streaming — batch upload order items
  rpc BatchCreateOrders(stream CreateOrderItemRequest) returns (BatchCreateOrderResponse);

  // Bidirectional streaming — live order stream with acknowledgments
  rpc StreamOrders(stream OrderEvent) returns (stream OrderAck);
}

Pattern 2: Go gRPC Server Implementation

Proper context handling, typed error responses, and interceptor patterns.

package server

import (
	"context"
	"fmt"
	"log/slog"
	"time"

	pb "github.com/myorg/orders/proto/gen/orders/v1"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
)

// Server implements the generated OrderServiceServer interface.
type Server struct {
	pb.UnimplementedOrderServiceServer
	orderStore   OrderStore
	logger       *slog.Logger
	maxStreamAge time.Duration
}

func NewServer(store OrderStore, logger *slog.Logger) *Server {
	return &Server{
		orderStore:   store,
		logger:       logger,
		maxStreamAge: 5 * time.Minute,
	}
}

// CreateOrder implements the unary RPC with full error handling.
func (s *Server) CreateOrder(ctx context.Context, req *pb.CreateOrderRequest) (*pb.CreateOrderResponse, error) {
	// Validate inputs early — return InvalidArgument for bad data
	if req.UserId == "" {
		return nil, status.Error(codes.InvalidArgument, "user_id is required")
	}
	if len(req.Items) == 0 {
		return nil, status.Error(codes.InvalidArgument, "at least one item is required")
	}
	for _, item := range req.Items {
		if item.ProductId == "" {
			return nil, status.Error(codes.InvalidArgument, "item product_id cannot be empty")
		}
		if item.Quantity <= 0 {
			return nil, status.Error(codes.InvalidArgument, fmt.Sprintf("item quantity must be positive, got %d", item.Quantity))
		}
	}

	// Business logic: create order in store
	orderID, err := s.orderStore.CreateOrder(ctx, req)
	if err != nil {
		// Map domain errors to gRPC status codes
		if IsDuplicateError(err) {
			return nil, status.Error(codes.AlreadyExists, fmt.Sprintf("order already exists for user %s", req.UserId))
		}
		if IsNotFoundError(err) {
			return nil, status.Error(codes.NotFound, fmt.Sprintf("product not found: %v", err))
		}
		s.logger.ErrorContext(ctx, "failed to create order", "error", err)
		return nil, status.Errorf(codes.Internal, "internal error creating order")
	}

	return &pb.CreateOrderResponse{
		OrderId:   orderID,
		CreatedAt: timestamppb.Now(),
		Status:    pb.OrderStatus_ORDER_STATUS_PENDING,
	}, nil
}

// TrackOrder implements server-side streaming with context cancellation.
func (s *Server) TrackOrder(req *pb.TrackOrderRequest, stream pb.OrderService_TrackOrderServer) error {
	if req.GetOrderId() == "" {
		return status.Error(codes.InvalidArgument, "order_id is required")
	}

	ctx := stream.Context()
	streamTicker := time.NewTicker(2 * time.Second)
	defer streamTicker.Stop()

	for {
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-streamTicker.C:
			statusUpdate, err := s.orderStore.GetOrderStatus(ctx, req.OrderId)
			if err != nil {
				if errors.Is(err, ErrNotFound) {
					return status.Error(codes.NotFound, fmt.Sprintf("order %s not found", req.OrderId))
				}
				s.logger.ErrorContext(ctx, "failed to get order status", "error", err)
				return status.Errorf(codes.Internal, "stream error: %v", err)
			}

			update := &pb.OrderStatusUpdate{
				OrderId:  req.GetOrderId(),
				Status:   statusUpdate.Status,
				Updated:  timestamppb.Now(),
				Message:  statusUpdate.Message,
			}

			if err := stream.Send(update); err != nil {
				return status.Errorf(codes.Internal, "failed to send update: %v", err)
			}
		}
	}
}

Pattern 3: Go Interceptor Middleware Chain

Server-side unary interceptor for auth and logging.

// AuthInterceptor validates authentication tokens on every gRPC call.
func AuthInterceptor() grpc.UnaryServerInterceptor {
	return func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
		start := time.Now()

		// Extract token from metadata
		md, ok := metadata.FromIncomingContext(ctx)
		if !ok {
			return nil, status.Error(codes.Unauthenticated, "missing metadata")
		}

		tokens := md.Get("authorization")
		if len(tokens) == 0 || tokens[0] == "" {
			return nil, status.Error(codes.Unauthenticated, "authorization header required")
		}

		// Validate the JWT token
		claims, err := validateJWT(tokens[0])
		if err != nil {
			return nil, status.Errorf(codes.Unauthenticated, "invalid token: %v", err)
		}

		// Pass validated claims to downstream handlers via context
		ctx = context.WithValue(ctx, authClaimsKey{}, claims)

		// Call the actual handler
		response, err := handler(ctx, req)
		if err != nil {
			st := status.Convert(err)
			logRPC(info.FullMethod, time.Since(start), st.Code(), true)
			return nil, err
		}

		logRPC(info.FullMethod, time.Since(start), 0, false)
		return response, nil
	}
}

// LoggingInterceptor logs every RPC with method, duration, and status code.
func LoggingInterceptor(logger *slog.Logger) grpc.UnaryServerInterceptor {
	return func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
		start := time.Now()
		response, err := handler(ctx, req)

		statusCode := codes.OK
		if st, ok := status.FromError(err); ok {
			statusCode = st.Code()
		}

		fields := []any{
			"method", info.FullMethod,
			"duration_ms", time.Since(start).Milliseconds(),
			"status", statusCode.String(),
			"error", err,
		}

		if statusCode != codes.OK {
			logger.Error("rpc failed", fields...)
		} else {
			logger.Debug("rpc completed", fields...)
		}

		return response, err
	}
}

Pattern 4: Python gRPC Implementation

Complete service implementation with async client and error handling.

# server.py — gRPC service implementation
from concurrent import futures
import logging
from datetime import datetime
from google.protobuf import timestamp_pb2 as ts
import grpc
from grpc import StatusCode
from proto.gen.orders.v1 import orders_pb2
from proto.gen.orders.v1 import orders_pb2_grpc

logger = logging.getLogger(__name__)


class OrderService(orders_pb2_grpc.OrderServiceServicer):
    """Production gRPC service with typed error handling."""

    def __init__(self, order_store: OrderStore) -> None:
        self._store = order_store
        self._stream_interval = 2.0  # seconds between status updates

    async def CreateOrder(
        self, request: orders_pb2.CreateOrderRequest, context: grpc.ServicerContext
    ) -> orders_pb2.CreateOrderResponse:
        # Input validation
        if not request.user_id:
            context.set_code(StatusCode.INVALID_ARGUMENT)
            context.set_details("user_id is required")
            return orders_pb2.CreateOrderResponse()

        if not request.items:
            context.set_code(StatusCode.INVALID_ARGUMENT)
            context.set_details("at least one item is required")
            return orders_pb2.CreateOrderResponse()

        for i, item in enumerate(request.items):
            if not item.product_id:
                context.set_code(StatusCode.INVALID_ARGUMENT)
                context.set_details(f"item {i} product_id cannot be empty")
                return orders_pb2.CreateOrderResponse()
            if item.quantity <= 0:
                context.set_code(StatusCode.INVALID_ARGUMENT)
                context.set_details(f"item {i} quantity must be positive")
                return orders_pb2.CreateOrderResponse()

        try:
            order_id = await self._store.create_order(request)
        except DuplicateOrderError as e:
            context.set_code(StatusCode.ALREADY_EXISTS)
            context.set_details(str(e))
            return orders_pb2.CreateOrderResponse()
        except ProductNotFoundError as e:
            context.set_code(StatusCode.NOT_FOUND)
            context.set_details(str(e))
            return orders_pb2.CreateOrderResponse()

        now = ts.Timestamp()
        now.FromDatetime(datetime.utcnow())

        return orders_pb2.CreateOrderResponse(
            order_id=order_id,
            created_at=now,
            status=orders_pb2.OrderStatus.ORDER_STATUS_PENDING,
        )

    async def TrackOrder(
        self,
        request: orders_pb2.TrackOrderRequest,
        context: grpc.ServicerContext,
    ) -> grpc.AsyncGenerator[orders_pb2.OrderStatusUpdate, None]:
        """Server-streaming RPC with proper cancellation handling."""
        if not request.order_id:
            context.set_code(StatusCode.INVALID_ARGUMENT)
            context.set_details("order_id is required")
            return

        while True:
            # Check if client has cancelled
            if context.is_cancelled():
                break

            try:
                status_update = await self._store.get_order_status(request.order_id)
                update_msg = orders_pb2.OrderStatusUpdate(
                    order_id=request.order_id,
                    status=status_update.status,
                    message=status_update.message or "",
                )
                now = ts.Timestamp()
                now.FromDatetime(datetime.utcnow())
                update_msg.updated.CopyFrom(now)
                yield update_msg
            except OrderNotFoundError:
                context.set_code(StatusCode.NOT_FOUND)
                context.set_details(f"order {request.order_id} not found")
                return

            await asyncio.sleep(self._stream_interval)


def serve() -> None:
    """Configure and start the gRPC server with interceptors."""
    interceptors = [
        auth_interceptor(),       # Auth first
        logging_interceptor(),    # Logging second
        timeout_interceptor(10),  # Timeout last
    ]

    server = grpc.aio.server(
        futures.ThreadPoolExecutor(max_workers=50),
        interceptors=interceptors,
        options=[
            ("grpc.max_send_message_length", 50 * 1024 * 1024),  # 50MB
            ("grpc.max_receive_message_length", 50 * 1024 * 1024),
        ],
    )

    orders_pb2_grpc.add_OrderServiceServicer_to_server(
        OrderService(order_store=OrderStore()), server
    )

    server.add_insecure_port("[::]:50051")
    logger.info("Starting gRPC server on :50051")
    server.start()
    server.wait_for_termination()

Pattern 5: Python Client with Retry and Interceptors

Production client with exponential backoff retry for transient failures.

# client.py — gRPC client with retry, metrics, and proper error handling
import asyncio
import logging
from datetime import datetime
from proto.gen.orders.v1 import orders_pb2, orders_pb2_grpc
import grpc
from grpc import StatusCode

logger = logging.getLogger(__name__)


class OrderClient:
    """Production-ready gRPC client with retry logic and interceptors."""

    RETRYABLE_CODES = {StatusCode.UNAVAILABLE, StatusCode.DEADLINE_EXCEEDED}
    MAX_RETRIES = 3
    INITIAL_BACKOFF = 0.1  # seconds
    MAX_BACKOFF = 5.0

    def __init__(self, target: str = "localhost:50051", deadline: float = 10.0) -> None:
        self._target = target
        self._deadline = deadline
        self._channel: grpc.aio.Channel | None = None
        self._stub: orders_pb2_grpc.OrderServiceStub | None = None

    async def __aenter__(self) -> "OrderClient":
        await self.connect()
        return self

    async def __aexit__(self, *args: object) -> None:
        await self.close()

    async def connect(self) -> None:
        """Create channel and stub with proper options."""
        self._channel = grpc.aio.insecure_channel(
            self._target,
            options=[
                ("grpc.keepalive_time_ms", 30_000),
                ("grpc.keepalive_timeout_ms", 10_000),
                ("grpc.max_send_message_length", 50 * 1024 * 1024),
                ("grpc.max_receive_message_length", 50 * 1024 * 1024),
            ],
        )
        self._stub = orders_pb2_grpc.OrderServiceStub(self._channel)

    async def close(self) -> None:
        if self._channel:
            await self._channel.close()

    async def create_order(
        self, user_id: str, items: list[dict[str, int | float]], shipping_address_id: str
    ) -> tuple[str, datetime]:
        """Create an order with retry for transient failures."""
        if not user_id:
            raise ValueError("user_id is required")

        for i, item in enumerate(items):
            if item.get("quantity", 0) <= 0:
                raise ValueError(f"item {i} quantity must be positive")

        request = orders_pb2.CreateOrderRequest(
            user_id=user_id,
            items=[
                orders_pb2.OrderItem(
                    product_id=it["product_id"],
                    quantity=it["quantity"],
                    unit_price_cents=int(it.get("unit_price_cents", 0)),
                )
                for it in items
            ],
            shipping_address_id=shipping_address_id,
        )

        last_error: Exception | None = None
        backoff = self.INITIAL_BACKOFF

        for attempt in range(self.MAX_RETRIES):
            try:
                response = await asyncio.wait_for(
                    self._stub.CreateOrder(request),
                    timeout=self._deadline,
                )
                return (
                    response.order_id,
                    datetime.fromtimestamp(
                        response.created_at.seconds + response.created_at.nanos / 1e9
                    ),
                )
            except grpc.aio.AioRpcError as e:
                if e.code() not in self.RETRYABLE_CODES or attempt == self.MAX_RETRIES - 1:
                    raise

                last_error = e
                logger.warning(
                    "Retryable error on attempt %d/%d: %s",
                    attempt + 1, self.MAX_RETRIES, e.details(),
                )
                await asyncio.sleep(backoff)
                backoff = min(backoff * 2, self.MAX_BACKOFF)

        raise last_error or RuntimeError("create_order failed after retries")

    async def track_order(self, order_id: str) -> None:
        """Stream order status updates with cancellation support."""
        if not self._stub:
            raise RuntimeError("client not connected")

        request = orders_pb2.TrackOrderRequest(order_id=order_id)
        deadline = asyncio.get_event_loop().time() + self._deadline

        async for update in self._stub.TrackOrder(request):
            logger.info(
                "Order %s status: %s — %s",
                order_id, update.status.name, update.message,
            )
            # Client cancels after deadline
            if asyncio.get_event_loop().time() > deadline:
                break


# Usage example
async def main() -> None:
    async with OrderClient(target="orders-service:50051", deadline=10.0) as client:
        order_id, created = await client.create_order(
            user_id="usr-123",
            items=[{"product_id": "prod-456", "quantity": 2, "unit_price_cents": 1999}],
            shipping_address_id="addr-789",
        )
        print(f"Order {order_id} created at {created}")

        await client.track_order(order_id)

Pattern 6: Bidirectional Streaming with Flow Control

Bidirectional streaming requires careful flow control and backpressure handling.

// StreamOrders handles bidirectional streaming between client and server.
// Each side sends independent message streams that must be read concurrently.
func (s *Server) StreamOrders(stream pb.OrderService_StreamOrdersServer) error {
	ctx := stream.Context()
	done := make(chan struct{})

	// Worker: read from client and process events
	go func() {
		defer close(done)
		for {
			event, err := stream.Recv()
			if err == io.EOF {
				return
			}
			if err != nil {
				s.logger.ErrorContext(ctx, "stream recv error", "error", err)
				return
			}

			switch event.GetEventType() {
			case pb.OrderEvent_EVENT_TYPE_NEW_ORDER:
				statusUpdate, err := s.processNewOrder(ctx, event.GetNewOrder())
				if err != nil {
					_ = stream.Send(&pb.OrderAck{
						EventId: event.EventId,
						Result: &pb.OrderAck_Error_{
							Error: &pb.OrderAck_Error{
								Code:    pb.StatusError_INTERNAL_ERROR,
								Message: err.Error(),
							},
						},
					})
					continue
				}

				_ = stream.Send(&pb.OrderAck{
					EventId: event.EventId,
					Result: &pb.OrderAck_Success{
						Success: statusUpdate,
					},
				})

			case pb.OrderEvent_EVENT_TYPE_CANCEL_ORDER:
				_ = stream.Send(&pb.OrderAck{
					EventId: event.EventId,
					Result:  &pb.OrderAck_Cancel{Cancel: &pb.CancelAck{}},
				})
			}
		}
	}()

	// Main loop: handle server-sent notifications and client cancellation
	for {
		select {
		case <-ctx.Done():
			return ctx.Err()
		case notification := <-s.notificationChan:
			if err := stream.Send(notification); err != nil {
				s.logger.ErrorContext(ctx, "failed to send notification", "error", err)
				return status.Errorf(codes.Internal, "send error: %v", err)
			}
		case <-done:
			return nil
		}
	}
}

Pattern 7: gRPC Status Code Mapping Reference

Map every domain error to the correct gRPC status code. Using wrong codes causes confusion for clients implementing retry logic.

gRPC Status Code	When to Use	Example Scenario	Retry Strategy
OK	Call succeeded	Normal response	N/A
InvalidArgument	Client sent bad input	Missing required field, validation failed	Never retry — fix the client request
NotFound	Resource does not exist	User ID not found in database	Never retry
AlreadyExists	Duplicate creation	Creating user that already exists	Never retry
PermissionDenied	Authenticated but unauthorized	Insufficient permissions for action	Never retry
Unauthenticated	Missing or invalid auth	No token, expired JWT	Never retry
FailedPrecondition	Operation rejected due to state	Trying to cancel a delivered order	Never retry — fix the precondition
Aborted	Operation was aborted	Concurrent modification conflict	Retry once with backoff
OutOfRange	Value outside valid range	Negative quantity, date out of range	Never retry
Internal	Server-side unexpected error	Panics, unhandled exceptions	Retry with backoff (max 3 attempts)
Unavailable	Service temporarily down	Connection refused, upstream timeout	Retry aggressively with exponential backoff
DeadlineExceeded	Request timed out	Slow database query exceeded deadline	Retry once with longer deadline

---

Constraints

MUST DO

• Define the .proto contract before writing any service or client code
• Use versioned package names (package myapp.v1) to enable API evolution
• Never reuse field numbers in a message — once assigned, they are permanent
• Set context deadlines on every gRPC client call (never leave timeouts at infinity)
• Map every error path to a specific gRPC status code using status.Error() (Go) or context.set_code() (Python)
• Handle stream cancellation explicitly with ctx.Done() checks in all streaming methods
• Implement interceptors as composable units — each interceptor handles one concern

MUST NOT DO

• Use string-based error messages instead of gRPC status codes for client-facing errors
• Return raw panics or unhandled errors from service handlers (always catch and return status.Error)
• Leave streaming RPCs running without a context deadline or cancellation check
• Put sensitive data (passwords, tokens) in protobuf message fields — use metadata/metadata-only transport instead
• Share gRPC connections across unrelated request scopes — create per-service or use connection pooling
• Mix protobuf field usage between services — keep messages scoped to their own service package

---

Testing Patterns

Go Service Unit Test

func TestCreateOrder(t *testing.T) {
	store := &mockOrderStore{
		createFn: func(ctx context.Context, req *pb.CreateOrderRequest) (string, error) {
			return "ord-123", nil
		},
	}
	srv := NewServer(store, slog.Default())

	req := &pb.CreateOrderRequest{
		UserId:   "usr-456",
		Items:    []*pb.OrderItem{{ProductId: "prod-1", Quantity: 2}},
		ShippingAddressId: "addr-789",
	}

	resp, err := srv.CreateOrder(context.Background(), req)
	require.NoError(t, err)
	assert.Equal(t, "ord-123", resp.GetOrderId())
	assert.Equal(t, pb.OrderStatus_ORDER_STATUS_PENDING, resp.GetStatus())
}

func TestCreateOrderInvalidInput(t *testing.T) {
	srv := NewServer(nil, slog.Default())
	req := &pb.CreateOrderRequest{} // Missing required fields

	_, err := srv.CreateOrder(context.Background(), req)
	assert.Error(t, err)

	st := status.Convert(err)
	assert.Equal(t, codes.InvalidArgument, st.Code())
}

Python Service Test

# test_server.py — pytest for gRPC service
import pytest
from proto.gen.orders.v1 import orders_pb2
from server import OrderService


class MockOrderStore:
    async def create_order(self, request):
        if not request.user_id:
            raise ValueError("user_id required")
        return "ord-mock-456"

    async def get_order_status(self, order_id):
        return type("Status", (), {"status": 2, "message": "shipped"})()


@pytest.mark.asyncio
async def test_create_order_success():
    store = MockOrderStore()
    service = OrderService(store)

    request = orders_pb2.CreateOrderRequest(
        user_id="usr-123",
        items=[orders_pb2.OrderItem(product_id="prod-1", quantity=2, unit_price_cents=999)],
    )

    response = await service.CreateOrder(request, MockContext())
    assert response.order_id == "ord-mock-456"
    assert response.status == orders_pb2.OrderStatus.ORDER_STATUS_PENDING


@pytest.mark.asyncio
async def test_create_order_missing_user_id():
    store = MockOrderStore()
    service = OrderService(store)

    request = orders_pb2.CreateOrderRequest(user_id="")

    response = await service.CreateOrder(request, MockContext())
    assert response.order_id == ""


class MockContext:
    """Minimal ServicerContext mock for unit testing."""
    code = StatusCode.OK
    details = ""

    def set_code(self, code):
        self.code = code

    def set_details(self, details):
        self.details = details

---

Protobuf Best Practices

Type Selection Guidelines

Scenario	Recommended Type	Why
Unique identifiers (UUIDs, Snowflake IDs)	string with documented format	Human-readable, no overflow risk
Numeric IDs for lookups	int64 or uint64	Sufficient range for counter-based IDs
Monetary amounts	int64 in smallest unit (cents, satoshis)	Avoids float precision loss entirely
Timestamps	google.protobuf.Timestamp	Standardized, timezone-naive UTC
Duration	google.protobuf.Duration	Standardized duration type
Boolean flags	bool	Simple true/false state
Collections	repeated <type>	Ordered, allows zero or more elements
Enumerated states	enum with UNSPECIFIED = 0 sentinel	Type-safe state machine

Field Number Strategy

1. Reserve numbers for frequently-used fields (1–15) — these encode most efficiently
2. Use higher numbers (16+) for rarely-set optional fields
3. Reserve blocks of numbers for future expansion: reserve 17, 20, 25;
4. Never reassign a field number even if the original field is removed
5. When removing a field, mark it as deprecated and reserve the number

// ✅ GOOD — reserved blocks prevent accidental reuse
message User {
  string id = 1;           // Most common query field
  string email = 2;
  string name = 3;
  bool is_active = 4 [deprecated = true];  // Mark for removal, keep number

  reserved 5, 6, 7;        // Reserved for future use — blocks reuse
}

---

Code Generation Commands

Go

# Install protoc plugins
go install google.golang.org/protobuf/cmd/protoc-gen-go@latest
go install google.golang.org/grpc/cmd/protoc-gen-go-grpc@latest

# Generate code
protoc --proto_path=. \
  --go_out=./proto/gen \
  --go_opt=paths=source_relative \
  --go-grpc_out=./proto/gen \
  --go-grpc_opt=paths=source_relative \
  proto/orders/v1/orders.proto

Python

# Install protoc tools
pip install grpcio grpcio-tools protobuf

# Generate code
python -m grpc_tools.protoc \
  -I. \
  --python_out=./proto/gen \
  --grpc_python_out=./proto/gen \
  proto/orders/v1/orders.proto

Buf (Modern Alternative)

# buf.yaml — recommended for team-wide consistency
version: v1
breaking:
  use:
    - FILE
lint:
  use:
    - DEFAULT
build:
  roots:
    - proto

---

Related Skills

Skill	Purpose
grpc (cncf)	Kubernetes deployment, service mesh integration, and infrastructure configuration for gRPC services
rest-api-patterns	Design REST APIs for external-facing interfaces; use gRPC internally and REST externally via gateway
fastapi-patterns	Build Python HTTP/REST services alongside gRPC microservices in the same codebase

---

Troubleshooting

Connection Issues

• UNAVAILABLE: connection refused — Server not running or wrong port. Verify server is listening and firewall rules allow traffic on the target port.
• RESOURCE_EXHAUSTED: tried to send message larger than max — Message size exceeds default 4MB limit. Set grpc.max_send_message_length and grpc.max_receive_message_length options on both client and server.
• Connections dropping behind load balancers — Enable keepalive with keepalive.Time, keepalive.Timeout, and keepalive.PermitWithoutStream options.

Streaming Issues

• Stream hangs without sending data — Check that the stream sender is not blocked waiting for a receiver. Implement proper backpressure with buffered channels or bounded queues.
• Memory leaks from unread streams — Clients MUST drain or cancel streaming responses. Always use defer cancel() and check context cancellation in server-side streams.

Protobuf Issues

• "field number X reused" — Field numbers are permanent once assigned in a deployed schema. Use new field numbers for additions, mark old ones as deprecated before removal.
• Generated code doesn't compile — Verify protoc versions match between development and CI environments. Pin the protoc version in CI pipelines.