Межсервисное взаимодействие: gRPC, Protocol Buffers и Service Mesh

Сравнение накладных расходов REST/JSON и gRPC/Protobuf в проводном формате (Wire Format):

1. REST / JSON через HTTP/1.1:
[HTTP/1.1 Headers: ~500-800 байт текстовых метаданных CRLF]
{"userId": "usr_9921", "amount": 1500.50, "currency": "USD", "isExecuted": true}
- Размер сообщения: ~620 байт (включая заголовки)
- CPU оверхед: Парсинг строк, кодирование UTF-8, инспекция типов V8 runtime

2. gRPC / Protocol Buffers v3 через HTTP/2:
[HTTP/2 HPACK HEADERS Frame: ~12-25 байт сжатых индексов]
[5 байт gRPC Frame Header: 1 байт Compressed Flag + 4 байта Message Length]
[Protobuf Binary Payload: 18 байт]
08 eb 4d 11 00 00 00 00 00 d2 97 40 1a 03 55 53 44 20 01
- Размер сообщения: ~40 байт (сокращение трафика на 80-90%)
- CPU оверхед: Побитовые сдвиги и прямое чтение типизированных смещений без выделения строк

Key = (field_number << 3) | wire_type

4 типа коммуникационных паттернов в gRPC:

1. Unary RPC (Одиночный запрос - Одиночный ответ):
Client ──[Request: UserRequest]──> Server
Client <──[Response: UserProfile]── Server

2. Server Streaming RPC (Одиночный запрос - Поток ответов):
Client ──[Request: SubscribeEvents]──> Server
Client <──[Stream: EventChunk 1]────── Server
Client <──[Stream: EventChunk 2]────── Server
Client <──[Stream: EventChunk N]────── Server

3. Client Streaming RPC (Поток запросов - Одиночный ответ):
Client ──[Stream: MetricPoint 1]────> Server
Client ──[Stream: MetricPoint 2]────> Server
Client ──[Stream: MetricPoint N]────> Server
Client <──[Response: AggregationSummary]── Server

4. Bidirectional Streaming RPC (Двунаправленный независимый поток):
Client ──[Stream: AudioChunk 1]─────> Server
Client <──[Stream: SubtitleChunk 1]── Server
Client ──[Stream: AudioChunk 2]─────> Server
Client <──[Stream: SubtitleChunk 2]── Server

syntax = "proto3";

package billing.v1;

// Определение сервиса биллинга
service BillingService {
  // 1. Unary RPC: Создание платежа
  rpc ProcessPayment (PaymentRequest) returns (PaymentResponse);

  // 2. Server Streaming RPC: Получение выписки транзакций
  rpc StreamTransactions (TransactionFilter) returns (stream Transaction);

  // 3. Client Streaming RPC: Пакетная загрузка биллинговых метрик
  rpc IngestMetrics (stream MetricRecord) returns (IngestSummary);

  // 4. Bidirectional Streaming: Интерактивная сверка баланса в реальном времени
  rpc LiveReconciliation (stream BalanceAuditRequest) returns (stream BalanceAuditResponse);
}

// Зарезервированные теги и имена для безопасной эволюции схемы
message PaymentRequest {
  reserved 4, 10 to 15;
  reserved "internal_salt";

  string account_id = 1;
  double amount = 2;
  string currency = 3;
  PaymentMethod method = 5;
  map<string, string> metadata = 6;
}

enum PaymentMethod {
  PAYMENT_METHOD_UNSPECIFIED = 0;
  PAYMENT_METHOD_CREDIT_CARD = 1;
  PAYMENT_METHOD_BANK_TRANSFER = 2;
  PAYMENT_METHOD_CRYPTO = 3;
}

message PaymentResponse {
  string transaction_id = 1;
  PaymentStatus status = 2;
  int64 timestamp_epoch_ms = 3;
}

enum PaymentStatus {
  PAYMENT_STATUS_UNSPECIFIED = 0;
  PAYMENT_STATUS_SUCCESS = 1;
  PAYMENT_STATUS_DECLINED = 2;
  PAYMENT_STATUS_PENDING = 3;
}

message TransactionFilter {
  string account_id = 1;
  int64 from_timestamp = 2;
}

message Transaction {
  string id = 1;
  double amount = 2;
  string status = 3;
}

message MetricRecord {
  string metric_name = 1;
  double value = 2;
  int64 timestamp = 3;
}

message IngestSummary {
  int32 processed_count = 1;
  int64 elapsed_ms = 2;
}

message BalanceAuditRequest {
  string audit_session_id = 1;
  double reported_balance = 2;
}

message BalanceAuditResponse {
  bool is_matched = 1;
  double discrepancy = 2;
}

import * as grpc from "@grpc/grpc-js";
import * as protoLoader from "@grpc/proto-loader";
import path from "node:path";

// Загрузка контракта Protobuf
const PROTO_PATH = path.resolve(process.cwd(), "proto/billing.proto");
const packageDefinition = protoLoader.loadSync(PROTO_PATH, {
  keepCase: true,
  longs: String,
  enums: String,
  defaults: true,
  oneofs: true,
});

const protoDescriptor = grpc.loadPackageDefinition(packageDefinition) as any;
const billingPackage = protoDescriptor.billing.v1;

const server = new grpc.Server({
  // Тюнинг HTTP/2 параметров сетевого стека gRPC
  "grpc.max_receive_message_length": 4 * 1024 * 1024, // 4 MB
  "grpc.max_send_message_length": 4 * 1024 * 1024,
  "grpc.http2.max_pings_without_data": 0,
  "grpc.keepalive_time_ms": 30000,
  "grpc.keepalive_timeout_ms": 10000,
});

// 1. Unary Call: ProcessPayment
function processPayment(
  call: grpc.ServerUnaryCall<any, any>,
  callback: grpc.sendUnaryData<any>
) {
  const metadata = call.metadata;
  const traceId = metadata.get("x-trace-id")[0] || "no-trace";

  console.log(`[gRPC ProcessPayment] Trace: ${traceId}, Account: ${call.request.account_id}`);

  // Проверка наступления дедлайна (Deadline / Timeout)
  if (call.getDeadline() && new Date(call.getDeadline()).getTime() <= Date.now()) {
    return callback({
      code: grpc.status.DEADLINE_EXCEEDED,
      message: "Время ожидания выполнения запроса истекло на сервере",
    });
  }

  // Бизнес-логика обработки платежа
  const response = {
    transaction_id: "tx_" + Math.random().toString(36).substring(2, 11),
    status: "PAYMENT_STATUS_SUCCESS",
    timestamp_epoch_ms: Date.now(),
  };

  // Возврат ответа и метаданных трейлера
  const trailer = new grpc.Metadata();
  trailer.set("x-server-execution-time", "1.4ms");
  callback(null, response, trailer);
}

// 2. Server Streaming: StreamTransactions
function streamTransactions(call: grpc.ServerWritableStream<any, any>) {
  const { account_id } = call.request;
  console.log(`[gRPC Server Stream] Запрос транзакций для ${account_id}`);

  let count = 0;
  const interval = setInterval(() => {
    count++;
    call.write({
      id: `tx_${count}`,
      amount: Math.round(Math.random() * 500 * 100) / 100,
      status: "COMPLETED",
    });

    if (count >= 5) {
      clearInterval(interval);
      call.end(); // Завершение потока
    }
  }, 100);

  call.on("cancelled", () => {
    console.warn("[gRPC Server Stream] Клиент отменил вызов стрима");
    clearInterval(interval);
  });
}

// Регистрация сервиса в сервере
server.addService(billingPackage.BillingService.service, {
  ProcessPayment: processPayment,
  StreamTransactions: streamTransactions,
  IngestMetrics: (call: any, cb: any) => { /* Client stream handler */ },
  LiveReconciliation: (call: any) => { /* Bidi stream handler */ },
});

server.bindAsync(
  "0.0.0.0:50051",
  grpc.ServerCredentials.createInsecure(), // В продакшене: grpc.ServerCredentials.createSsl()
  (err, port) => {
    if (err) {
      console.error("[gRPC Server Bind Error]:", err);
      return;
    }
    console.log(`[gRPC Server] Запущен на порту ${port}`);
  }
);

import * as grpc from "@grpc/grpc-js";
import * as protoLoader from "@grpc/proto-loader";
import path from "node:path";

const PROTO_PATH = path.resolve(process.cwd(), "proto/billing.proto");
const packageDefinition = protoLoader.loadSync(PROTO_PATH);
const protoDescriptor = grpc.loadPackageDefinition(packageDefinition) as any;
const BillingService = protoDescriptor.billing.v1.BillingService;

// Interceptor для автоматического логирования и сквозной трассировки
const telemetryInterceptor: grpc.Interceptor = (options, nextCall) => {
  return new grpc.InterceptingCall(nextCall(options), {
    start: (metadata, listener, next) => {
      // Добавляем распределенный W3C Trace ID
      metadata.set("x-trace-id", "trace_" + Date.now());

      const customListener: grpc.Listener = {
        onReceiveMetadata: (receivedMetadata, nextMeta) => {
          nextMeta(receivedMetadata);
        },
        onReceiveMessage: (message, nextMsg) => {
          nextMsg(message);
        },
        onReceiveStatus: (status, nextStatus) => {
          if (status.code !== grpc.status.OK) {
            console.error(`[gRPC Error Status ${status.code}]: ${status.details}`);
          }
          nextStatus(status);
        },
      };

      next(metadata, customListener);
    },
  });
};

const client = new BillingService(
  "localhost:50051",
  grpc.credentials.createInsecure(),
  {
    interceptors: [telemetryInterceptor],
    "grpc.keepalive_time_ms": 15000,
  }
);

// Вызов Unary метода со строгим дедлайном (2000 мс)
async function executePayment() {
  const deadline = new Date();
  deadline.setMilliseconds(deadline.getMilliseconds() + 2000);

  const request = {
    account_id: "acc_77189",
    amount: 250.0,
    currency: "USD",
    method: "PAYMENT_METHOD_CREDIT_CARD",
    metadata: { orderId: "ord_100" },
  };

  client.ProcessPayment(
    request,
    { deadline },
    (error: grpc.ServiceError | null, response: any) => {
      if (error) {
        if (error.code === grpc.status.DEADLINE_EXCEEDED) {
          console.error("[gRPC Client] Запрос прерван: превышен дедлайн!");
        } else {
          console.error("[gRPC Client] Ошибка RPC:", error.message);
        }
        return;
      }
      console.log("[gRPC Client] Платеж успешно проведен:", response);
    }
  );
}

executePayment();