gRPC

Introduction

Featuregrpc gRPC gives a service a typed contract in a .proto file that any language can call. navree-grpc serves tonic services from your modules: you write the .proto, tonic generates the Rust types and the service trait, and you implement the trait on an #[injectable] struct marked #[grpc_service].

Navree runs its own pipeline around every call: the request id, authentication, guards, the access check, the deadline and a unit of work. A gRPC method declares who may call it with the same attributes as a route, takes the same extractors and returns the same AppError, which Navree turns into the matching gRPC status. The services run on your HTTP port by default, and answer gRPC, gRPC-Web and Connect clients.

Installation

navree-grpc is the grpc feature of the facade. The code tonic generates names tonic and prost directly, so your crate depends on them too, at the versions navree-grpc uses (cargo tree -p navree-grpc shows them). Code generation runs in build.rs with protox, which compiles .proto files in Rust, so you don't install protoc:

[dependencies]
navree = { git = "https://github.com/NavreeHQ/navree", features = ["grpc"] }
tonic = "0.14"
tonic-prost = "0.14"
prost = "0.14"
garde = { version = "0.23", features = ["derive"] } # rules on request messages
tokio-stream = { version = "0.1", features = ["sync"] } # streaming methods

[build-dependencies]
protox = "0.10"
tonic-prost-build = "0.14"
prost = "0.14"

The grpc feature isn't in the default full set, and it needs no system library.

Defining the service

Write the contract in a .proto file. This is the deals service's:

syntax = "proto3";

package crm.deals.v1;

service Deals {
  rpc GetDeal(GetDealRequest) returns (Deal);
  rpc ListDeals(ListDealsRequest) returns (ListDealsResponse);
  rpc MoveDeal(MoveDealRequest) returns (Deal);
  rpc WatchDeals(WatchDealsRequest) returns (stream DealEvent);
}

message GetDealRequest { string id = 1; }

message ListDealsRequest {
  string company_id = 1;
  uint32 page = 2;
  uint32 per_page = 3;
}

message ListDealsResponse {
  repeated Deal deals = 1;
  uint64 total = 2;
}

message MoveDealRequest {
  string id = 1;
  string stage = 2;
  optional int64 position = 3;
  optional string lost_reason = 4;
}

message Deal {
  string id = 1;
  string title = 2;
  int64 amount_cents = 3;
  string currency = 4;
  string stage = 5;
  string company_id = 6;
  string owner_id = 7;
}

message WatchDealsRequest {}

message DealEvent {
  string deal_id = 1;
  string stage = 2;
}

Generate the Rust code in build.rs. protox compiles the .proto files into a descriptor set, which you write to OUT_DIR for server reflection and Connect's JSON encoding, and tonic-prost-build generates the messages, the server trait and the client from it. The attributes add garde rules to the request messages you want validated:

use std::{env, fs, path::PathBuf};

use prost::Message;

fn main() -> Result<(), Box<dyn std::error::Error>> {
    println!("cargo:rerun-if-changed=proto");
    let descriptors = protox::compile(["crm/deals/v1/deals.proto"], ["proto"])?;
    let out = PathBuf::from(env::var("OUT_DIR")?);
    fs::write(out.join("deals_descriptor.bin"), descriptors.encode_to_vec())?;

    tonic_prost_build::configure()
        .type_attribute(".crm.deals.v1.MoveDealRequest", "#[derive(garde::Validate)]")
        .field_attribute(".crm.deals.v1.MoveDealRequest.id", "#[garde(skip)]")
        .field_attribute(".crm.deals.v1.MoveDealRequest.stage", "#[garde(length(min = 1))]")
        .field_attribute(".crm.deals.v1.MoveDealRequest.position", "#[garde(range(min = 0))]")
        .field_attribute(".crm.deals.v1.MoveDealRequest.lost_reason", "#[garde(length(chars, max = 1000))]")
        .compile_fds(descriptors)?;
    Ok(())
}

A garde derive needs a rule (or #[garde(skip)]) on every field of the message, so name each one.

Include the generated code in a module:

tonic::include_proto!("crm.deals.v1");

pub const FILE_DESCRIPTOR_SET: &[u8] = tonic::include_file_descriptor_set!("deals_descriptor");

Implementing it

Implement the generated trait on an injectable struct and mark the impl block #[grpc_service], naming the generated server type:

pub mod pb;

use crm_core::{Deal, Move};
use navree::grpc::{Request, Response, grpc_service};
use navree::http::CurrentPrincipal;
use navree::prelude::*;

use crate::service::DealService;
use pb::deals_server::{Deals, DealsServer};

#[injectable]
pub struct DealsGrpc {
    deals: Arc<DealService>,
}

#[grpc_service(server = DealsServer)]
#[authenticated]
impl Deals for DealsGrpc {
    #[can(Read, Deal, from = "id")]
    async fn get_deal(
        &self,
        _request: Request<pb::GetDealRequest>,
        uow: Uow,
        deal: Authorized<Deal, Read>,
    ) -> Result<Response<pb::Deal>> {
        let view = self.deals.view(&uow, deal.into_inner()).await?;
        Ok(Response::new(view.into()))
    }

    #[can(Read, Deal)]
    async fn list_deals(
        &self,
        request: Request<pb::ListDealsRequest>,
        uow: Uow,
        scope: Accessible<Deal, Read>,
    ) -> Result<Response<pb::ListDealsResponse>> {
        let input = request.into_inner();
        let page = self.deals.of_company_page(&uow, scope, &input).await?;
        Ok(Response::new(page.into()))
    }

    #[can(Move, Deal, from = "id")]
    async fn move_deal(
        &self,
        Valid(request): Valid<Request<pb::MoveDealRequest>>,
        uow: Uow,
        CurrentPrincipal(me): CurrentPrincipal,
        deal: Authorized<Deal, Move>,
    ) -> Result<Response<pb::Deal>> {
        let input = request.into_inner().try_into()?;
        let view = self.deals.move_to(&uow, me.user_id, deal.into_inner(), input).await?;
        Ok(Response::new(view.into()))
    }

    // `watch_deals`: see "Streaming" below.
}

The methods are tonic's, with three additions that #[grpc_service] handles:

• The first parameter is tonic's Request<T>, as the trait declares it, or Valid<Request<T>> to validate the message with its garde rules first.
• Navree extractors may follow it: Uow, CurrentPrincipal, CurrentUser, Abilities, Accessible<E, A>, Authorized<E, A>, Scoped<T>, CurrentWorkspace and your own (any FromRequestParts whose rejection is an AppError). The macro generates the method tonic expects and runs the extractors on the call's metadata and extensions.
• The result may be Result<Response<T>>, Navree's result with AppError, so ? works on database errors, a Status from another service and anything else. The macro converts errors to a Status (see errors). Returning Result<Response<T>, Status> works too. Write the result type out: the macro reads Response<T> from it, so a type alias for the whole result is an error.

The macro also adds tonic's #[async_trait] to the impl block, so you don't write it.

Conversions between your domain types and the generated messages are plain From and TryFrom impls, like impl From<DealView> for pb::Deal. A TryFrom that fails on a bad UUID or an unknown stage should return AppError::bad_request(..), which reaches the client as INVALID_ARGUMENT.

Registering the service

Register the struct in its module, with the descriptor set, and add GrpcModule to the app root:

#[module(providers = [DealService], register = Self::wire)]
pub struct DealsModule;

impl DealsModule {
    fn wire(&self, m: &mut ModuleBuilder) {
        m.grpc_service::<DealsGrpc>()
            .grpc_descriptors(grpc::pb::FILE_DESCRIPTOR_SET);
    }
}

platform(Navree::app(), ConfigSources::defaults())
    .module(GrpcModule::from_config())
    .module(CrmCoreModule)
    .module(DealsModule)
    .run()

m.grpc_service::<T>() registers T as a singleton of the module, built with dependency injection, and serves it. m.grpc_descriptors(..) hands the encoded descriptor set to reflection and to Connect's JSON encoding; a module may register several. Register GrpcModule after DatabaseModule (which platform adds), so the gRPC calls an HTTP handler makes carry its request's context.

The same DealService now answers HTTP routes and gRPC calls, with one set of access rules.

At boot, Navree checks every registered service and fails with the full list of problems when a method declares no access, is #[public] and restricted at once, needs an authenticator or a route authorizer the app doesn't have, or when two modules register the same service.

Serving

By default, gRPC shares the HTTP server's port. The server speaks HTTP/1.1 and HTTP/2 on one listener and sends requests whose content-type is application/grpc (or gRPC-Web or Connect, when they're on) to the gRPC services, and everything else to your routes. gRPC paths are absolute (/crm.deals.v1.Deals/GetDeal): http.api_prefix and route versions don't apply to them.

serve starts it all, and Ctrl-C or SIGTERM stops new calls, lets unary calls in flight finish, and ends open streams with UNAVAILABLE.

gRPC needs HTTP/2 end to end. Plain-text HTTP/2 (h2c) works in development and between services on a private network. Behind a reverse proxy, the proxy must speak HTTP/2 to the app: with Caddy, reverse_proxy h2c://127.0.0.1:8080, which is what Harbor configures when the app's Cargo.toml turns on navree's grpc feature or depends on navree-grpc.

To serve gRPC on a port of its own, set grpc.addr:

[grpc]
addr = "0.0.0.0:50051"
web = true
connect = true

The separate listener opens with serve, once the HTTP port is listening (other commands, such as migrate, never open it), speaks HTTP/1.1 and HTTP/2, and answers 415 to anything that isn't gRPC, gRPC-Web or Connect.

As environment variables: APP_GRPC__ADDR=0.0.0.0:50051, APP_GRPC__REFLECTION=false. In code: GrpcModule::for_root(GrpcOptions { addr: Some("0.0.0.0:50051".parse()?), ..GrpcOptions::default() }).

Inject GrpcServer to see what runs: services() lists the registered service names, and local_addr() (or wait_bound().await) gives the address of the grpc.addr listener, which helps with addr = "127.0.0.1:0" in tests.

The call pipeline

Every call, unary or streaming, goes through these steps before your method runs:

1. Request id and tracing. x-request-id from the metadata, or a new one, returned in the response metadata; the call runs in a grpc span with its path, request id, user and the caller's traceparent. Each finished call is logged with its code and latency, and counted in the grpc_server_handled_total and grpc_server_handling_seconds metrics.
2. The deadline: the client's grpc-timeout, or grpc.timeout_secs, counted from the call's arrival.
3. Interceptors you attached to the service, if any.
4. Authentication of the authorization metadata, by the app's dyn Authenticator: a bearer token works exactly as on HTTP. Invalid credentials end the call with UNAUTHENTICATED, and so does a missing one on a method that isn't #[public].
5. Guards (#[guard(..)] on the impl block or the method), which see the metadata as request headers.
6. The access check: #[roles], #[permissions] or #[can].
7. The unit of work, around the extractors and your method.

Access and guards

gRPC methods are deny by default: each needs an access attribute (#[public], #[authenticated], #[roles(..)], #[permissions(..)] or #[can(..)]), on the method or on the impl block, or the service doesn't compile. #[can(.., from = "id")] reads the record's id from the request message's id field (a string or an integer, optional or not), loads the record in the call's unit of work and checks it, so get_deal above answers NOT_FOUND or PERMISSION_DENIED before it runs, and INVALID_ARGUMENT when the id is empty. It needs a unary request: a method that receives a stream can't use from. See access control for the attributes.

Guards are the same Guard trait as on routes. A guard reading x-api-version from a route's headers reads it from the call's metadata here, unchanged. A refused call is recorded as access.denied in the audit log, with the gRPC path as the route.

The route attributes that only mean something for HTTP (#[get] and the other verbs, #[throttle], #[timeout], #[audit], #[name], #[api], #[transactional]) are compile errors on a gRPC method.

Validation

Valid<Request<T>> runs the garde rules of a message type, which you add in build.rs with type_attribute and field_attribute as above, with the app's validation messages. A message that breaks a rule ends the call with INVALID_ARGUMENT, a message listing every broken rule, and a google.rpc.BadRequest detail with one field violation per rule, which gRPC clients in every language can read:

code: INVALID_ARGUMENT
message: position must not be less than 0
details: google.rpc.BadRequest { field_violations: [ { field: "position", description: "position must not be less than 0" } ] }

The unit of work

A call gets a unit of work like an HTTP request: it commits when the method returns Ok, then the response is sent; it rolls back on an error or when the deadline passes. Jobs, notifications and events emitted during the call join it. For a unary or client-streaming call, the unit of work covers the whole method. For a server-streaming or bidirectional call, it covers the part of the method that opens the stream: the items the stream sends later run outside it, since a stream can last for hours. Open a transaction per item with Db::transaction when a stream writes.

Interceptors

Tonic interceptors work unchanged. Attach one to a service with the interceptor option; it runs before authentication, so use it for cheap checks on the raw metadata. A function like this one or any value that implements tonic's Interceptor works:

use navree::grpc::{Request, Status};

fn require_client_version(request: Request<()>) -> Result<Request<()>, Status> {
    match request.metadata().get("x-client-version") {
        Some(_) => Ok(request),
        None => Err(Status::failed_precondition("x-client-version is required")),
    }
}

#[grpc_service(server = DealsServer, interceptor = require_client_version)]
#[authenticated]
impl Deals for DealsGrpc {
    // ...
}

Errors

An AppError becomes the gRPC status that means the same thing, with the error's message (navree::grpc::status_from_error):

Server errors keep the message "Internal server error" on the wire and are logged with the request id, as on HTTP, and the app's error mappers (m.map_error::<E>(..)) apply to them first. A Status you return yourself is sent as it is.

The opposite mapping (navree::grpc::error_from_status) applies when your code calls another gRPC service: ? on a Status gives the AppError of the same meaning, so a NOT_FOUND from the deals service is a 404 in the API that called it, and an INVALID_ARGUMENT with field violations is a validation error with the same fields. UNIMPLEMENTED, UNAVAILABLE, DEADLINE_EXCEEDED and the client errors keep the upstream message; INTERNAL, UNKNOWN and the other server errors become a 500 that keeps it only in the logs.

Streaming

Server-streaming, client-streaming and bidirectional methods work as tonic defines them, and a method that receives a stream takes Request<Streaming<T>> (navree::grpc::Streaming). The access check and the guards run once, when the stream opens. This one streams deal changes from the app's event bus:

use std::pin::Pin;

use navree::events::EventBus;
use navree::grpc::Status;
use tokio_stream::{Stream, StreamExt, wrappers::BroadcastStream};

type DealEvents = Pin<Box<dyn Stream<Item = Result<pb::DealEvent, Status>> + Send>>;

#[grpc_service(server = DealsServer)]
#[authenticated]
impl Deals for DealsGrpc {
    type WatchDealsStream = DealEvents;

    #[can(Read, Deal)]
    async fn watch_deals(
        &self,
        _request: Request<pb::WatchDealsRequest>,
    ) -> Result<Response<Self::WatchDealsStream>> {
        let updates = BroadcastStream::new(self.events.subscribe_json("deal.updated"));
        let stream = updates.filter_map(|update| {
            let value = update.ok()?;
            Some(Ok(pb::DealEvent {
                deal_id: value["deal_id"].as_str()?.to_owned(),
                stage: value["stage"].as_str()?.to_owned(),
            }))
        });
        Ok(Response::new(Box::pin(stream) as DealEvents))
    }

    // ...
}

(events: Arc<EventBus> is a field of DealsGrpc here, and tokio-stream a dependency with its sync feature.) With several instances, each instance's bus only sees its own events; to see events published on every instance, broadcast them between instances.

When the app shuts down, open streams end with UNAVAILABLE; when the client's grpc-timeout passes, with DEADLINE_EXCEEDED.

Deadlines and metadata

A client's grpc-timeout is the call's deadline. When it passes, the method is cut off, its unit of work rolls back, and the client gets DEADLINE_EXCEEDED. Calls without one get grpc.timeout_secs. Calls your method makes through a Navree channel inherit the remaining time.

Navree reads and writes these metadata keys:

Read any other metadata from the request: request.metadata().get("x-client-version").

gRPC-Web and Connect

Browsers can't speak gRPC directly. With grpc.web and grpc.connect on (the default), the same services answer:

• gRPC-Web, binary and text, for clients generated by protoc-gen-grpc-web and Connect's gRPC-Web transport, over HTTP/1.1 or HTTP/2;
• Connect, a protocol of plain HTTP POSTs that carries protobuf or JSON, so a unary call is one request you can make with curl:

curl -X POST http://localhost:8081/crm.deals.v1.Deals/GetDeal \
  -H "content-type: application/json" \
  -H "authorization: Bearer $TOKEN" \
  -d '{"id": "0199c1d2-6f7e-7a31-b5d4-2f0c8e9a1b23"}'

{
  "id": "0199c1d2-6f7e-7a31-b5d4-2f0c8e9a1b23",
  "title": "Acme renewal",
  "amountCents": "1200000",
  "currency": "USD",
  "stage": "negotiation",
  "companyId": "0199c1d0-1b2c-7d3e-8f40-5a6b7c8d9e0f",
  "ownerId": "0192f0c1-0000-7000-8000-00000000000a"
}

JSON follows protobuf's JSON mapping (camelCase names, 64-bit integers as strings), transcoded with the descriptor sets registered with m.grpc_descriptors(..); a JSON call to a service without them answers UNIMPLEMENTED. Protobuf calls (application/proto) need no descriptors. Streaming methods answer Connect's streaming protocol (application/connect+proto and application/connect+json), and connect-timeout-ms sets a call's deadline. Errors use Connect's HTTP statuses and JSON error body. Connect calls are POSTs without compression, the default of Connect's clients.

On the shared port, a Connect request reaches the gRPC services only when its path names a service the app serves, so your JSON routes are never in the way. Browser calls go through the app's CORS settings (http.cors.allowed_origins); Navree adds the gRPC-Web and Connect headers to the allowed and exposed headers itself.

Reflection and health

In debug builds, server reflection is on, so tools discover your services without the .proto files:

grpcurl -plaintext localhost:8081 list
grpcurl -plaintext -H "authorization: Bearer $TOKEN" \
  -d '{"id": "0199c1d2-6f7e-7a31-b5d4-2f0c8e9a1b23"}' \
  localhost:8081 crm.deals.v1.Deals/GetDeal

grpc.health.v1.Health answers SERVING while the app's readiness checks pass and NOT_SERVING otherwise, for the server as a whole (the empty service name) and for each registered service, for load balancers and orchestrators that probe over gRPC; Watch sends the status again whenever it changes. Both are public; turn reflection off in production with grpc.reflection = false (the default in release builds) if your contract isn't public.

Calling other gRPC services

Name the services your app calls in grpc.clients:

[grpc.clients.deals]
url = "http://deals.internal:8081"
timeout_ms = 3000

Inject GrpcChannels and build tonic's generated client on the named channel:

use navree::grpc::GrpcChannels;
use crm_proto::deals::v1::{GetDealRequest, deals_client::DealsClient};

#[injectable]
pub struct DealLookup {
    channels: Arc<GrpcChannels>,
}

impl DealLookup {
    pub async fn get(&self, id: Uuid) -> Result<crm_proto::deals::v1::Deal> {
        let mut deals = DealsClient::new(self.channels.get("deals")?);
        let reply = deals.get_deal(GetDealRequest { id: id.to_string() }).await?;
        Ok(reply.into_inner())
    }
}

channels.get(name) is a 500 naming the missing key when grpc.clients.<name> isn't configured. Channels connect lazily, reconnect on their own and are shared by the whole app. A call on a Navree channel made while the app handles a gRPC call or an HTTP request carries that request's id, its trace context, its authorization (unless forward_auth is off), its workspace, and a grpc-timeout no later than the current deadline. Metadata you set yourself wins. Here the generated code comes from a crm-proto crate that both services depend on, which is the gRPC version of the shared contract crate.

Testing

navree::grpc::testing::TestChannel connects tonic clients to a TestApp in-process, with the whole pipeline and no port:

use deals::grpc::pb::{GetDealRequest, deals_client::DealsClient};
use navree::grpc::testing::TestChannel;
use navree::grpc::Code;
use navree::prelude::*;
use navree::testing::{TestApp, principal};
use uuid::Uuid;

#[tokio::test]
async fn get_deal_checks_access() {
    let app = TestApp::builder()
        .temp_database()
        .module(DatabaseModule::from_config())
        .module(AccessModule::for_root(crm_core::abilities::access_options()))
        .module(GrpcModule::from_config())
        .module(DealsModule)
        .build()
        .await;
    let (rep, deal_id) = seed_rep_with_deal(&app).await;
    let request = || GetDealRequest { id: deal_id.to_string() };

    let mut guest = DealsClient::new(TestChannel::new(&app));
    let status = guest.get_deal(request()).await.unwrap_err();
    assert_eq!(status.code(), Code::Unauthenticated);

    let mut stranger = DealsClient::new(TestChannel::new(&app).acting_as(principal(Uuid::now_v7())));
    let status = stranger.get_deal(request()).await.unwrap_err();
    assert_eq!(status.code(), Code::PermissionDenied);

    let mut owner = DealsClient::new(TestChannel::new(&app).acting_as(principal(rep)));
    let deal = owner.get_deal(request()).await.unwrap().into_inner();
    assert_eq!(deal.owner_id, rep.to_string());

    app.shutdown().await.unwrap();
}

acting_as signs calls in through the test app's TestAuthenticator, like TestApp::acting_as for HTTP requests. TestChannel::new(&app).bearer(token) sends real credentials instead, and .metadata(key, value) adds metadata (.metadata("x-workspace", "acme") names a workspace). To test a service that calls others over gRPC, register a stub implementation of their trait in a second test app with http.addr = "127.0.0.1:0", run its serve command (app.run_command("serve", &[])), and point grpc.clients.<name>.url at the address HttpServer::wait_bound() returns.