gRPC Firestore `Listen` in Rust

- 4 minutes read - 765 words

Obsessing on gRPC Firestore Listen somewhat but it’s also a good learning opportunity for me to write stuff in Rust. This doesn’t work against Google’s public endpoint (possibly for the same reason that gRPCurl doesn’t work either) but this does work against the Go server described in the other post.

I’m also documenting here because I always encounter challenges using TLS with Rust (and this documents 2 working ways to do this with gRPC) as well as references two interesting (rust) examples that use Google services.

Apologies in advance that the code would benefit from refactoring but I find it easier to chunk out a single function (main) when I’m testing.

Cargo.toml:

[package]
name = "rust"
version = "0.1.0"
edition = "2024"

[dependencies]
prost = "0.13.5"
prost-types = "0.13.5"
tonic = { version = "0.12.3", features = ["tls-native-roots"] }
tokio = { version = "1.43.0", features = ["full"] }
tokio-stream = "0.1.17"

[build-dependencies]
tonic-build = "0.12.3"

NOTE tonic’s tls-native-roots features

build.rs:

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

const GOOGLEAPIS: &str = "/path/to/googleapis";

fn main() -> Result<(), Box<dyn std::error::Error>> {
    // Required by prost-reflection to create firestore.binpb in correct OUT_DIR
    let out_dir = PathBuf::from(env::var("OUT_DIR").expect("unable to determine `OUT_DIR`"));

    tonic_build::configure()
        .build_client(true)
        .build_server(true)
        .file_descriptor_set_path(out_dir.join("firestore.binpb"))
        .compile_protos(
            &[format!("{GOOGLEAPIS}/google/firestore/v1/firestore.proto")],
            &[GOOGLEAPIS],
        )?;

    Ok(())
}

NOTE Uses a clone of Google’s gRPC GitHub repo called googleapis

Then:

main.rs:

pub mod google {
    pub mod firestore {
        pub mod v1 {
            tonic::include_proto!("google.firestore.v1");
        }
    }
    pub mod rpc {
        tonic::include_proto!("google.rpc");
    }
    pub mod r#type {
        tonic::include_proto!("google.r#type");
    }
}

use google::firestore::v1::{
    ListenRequest, StructuredQuery, Target, firestore_client::FirestoreClient,
    listen_request::TargetChange, structured_query::CollectionSelector, target::QueryTarget,
    target::TargetType, target::query_target::QueryType,
};
use std::env;
use tokio_stream::StreamExt;
use tonic::{
    metadata::MetadataMap,
    transport::{Channel, ClientTlsConfig},
};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let collection = env::var("COLLECTION")?;
    let endpoint = env::var("ENDPOINT")?;
    let project = env::var("PROJECT")?;
    let token = env::var("TOKEN")?;

    let uri: tonic::transport::Uri = endpoint.parse()?;

    // Uses the host's trusted certs
    let tls_config = ClientTlsConfig::new().with_native_roots();

    let channel = Channel::builder(uri)
        .tls_config(tls_config)?
        .connect()
        .await?;

    let bearer_token = format!("Bearer {token}");

    let database = format!("projects/{project}/databases/(default)");
    let parent = format!("{database}/documents");

    let mut metadata = MetadataMap::new();
    metadata.insert("authorization", bearer_token.parse()?);
    metadata.insert("google-cloud-resource-prefix", database.parse()?);

    let rqst = ListenRequest {
        database: database,
        labels: Default::default(),
        target_change: Some(TargetChange::AddTarget(Target {
            target_id: 0,
            once: false,
            expected_count: None,
            target_type: Some(TargetType::Query(QueryTarget {
                parent: parent,
                query_type: Some(QueryType::StructuredQuery(StructuredQuery {
                    select: None,
                    from: vec![CollectionSelector {
                        collection_id: collection,
                        all_descendants: false,
                    }],
                    r#where: None,
                    order_by: vec![],
                    start_at: None,
                    end_at: None,
                    offset: 0,
                    limit: None,
                })),
            })),
            resume_type: None,
        })),
    };

    let stream = tokio_stream::iter(vec![rqst]);
    let mut rqst = tonic::Request::new(stream);
    *rqst.metadata_mut() = metadata;

    let mut client = FirestoreClient::new(channel);
    let mut stream = client.listen(rqst).await?.into_inner();
    while let Some(resp) = stream.next().await {
        match resp {
            Ok(resp) => {
                println!("{:?}", resp);
            }
            Err(e) => {
                println!("{:?}", e);
            }
        }
    }

    Ok(())
}

NOTE

  1. The code would benefit from a renewing token source but, for now, TOKEN=$(gcloud auth print-access-token)
  2. with_native_certs is critical albeit slightly obscure and saves creating/loading a CA PEM

I’m using tailscale serve to get a TLS-enabled endpoint to the local Go service. With the above code:

HOST="..."
TAILNET="..."
ENDPOINT="${HOST}.${TAILNET}"

TOKEN=$(gcloud auth print-access-token) \
ENDPOINT="https://${ENDPOINT}" \
PROJECT="PROJECT" \
COLLECTION="collection" \
cargo run

Yields:

ListenResponse { response_type: Some(DocumentChange(DocumentChange { document: Some(Document { name: "d-1739487512", fields: {}, create_time: Some(Timestamp { seconds: 1739487512, nanos: 963638476 }), update_time: Some(Timestamp { seconds: 1739487512, nanos: 963638476 }) }), target_ids: [0], removed_target_ids: [] })) }
ListenResponse { response_type: Some(DocumentChange(DocumentChange { document: Some(Document { name: "d-1739487517", fields: {}, create_time: Some(Timestamp { seconds: 1739487517, nanos: 964152089 }), update_time: Some(Timestamp { seconds: 1739487517, nanos: 964152089 }) }), target_ids: [0], removed_target_ids: [] })) }
ListenResponse { response_type: Some(DocumentChange(DocumentChange { document: Some(Document { name: "d-1739487522", fields: {}, create_time: Some(Timestamp { seconds: 1739487522, nanos: 965279940 }), update_time: Some(Timestamp { seconds: 1739487522, nanos: 965279940 }) }), target_ids: [0], removed_target_ids: [] })) }

Replacing ENDPOINT="https://firestore.googleapis.com", a corrected PROJECT and COLLECTION yields no errors and no results.

An alternative to with_native_certs requires obtaining a CA PEM and loading this into the client code:

openssl s_client -showcerts -connect ${ENDPOINT}:443

Yields:

depth=2 C=US, O=Internet Security Research Group, CN=ISRG Root X1
verify return:1
depth=1 C=US, O=Let's Encrypt, CN=E6
verify return:1
depth=0 CN={HOST}
verify return:1
---
Certificate chain
 0 s:CN={HOST}
   i:C=US, O=Let's Encrypt, CN=E6
   a:PKEY: id-ecPublicKey, 256 (bit); sigalg: ecdsa-with-SHA384
   v:NotBefore: Feb 13 00:00:00 2025 GMT; NotAfter: Apr 13 23:59:59 2025 GMT
-----BEGIN CERTIFICATE-----
-----END CERTIFICATE-----

Using Let’s Encrypt’s Intermediate CAs, I captured the value from the Certificate details pem link:

Let’s Encrypt E6
Subject: O = Let's Encrypt, CN = E6
Key type: ECDSA P-384
Validity: until 2027-03-12
CA details: crt.sh, issued certs
Certificate details (signed by ISRG Root X2): der, pem, txt
Certificate details (cross-signed by ISRG Root X1): der, pem, txt

And then:

let pem = std::fs::read_to_string("ca.pem")?;
let ca = Certificate::from_pem(pem);
let tls_config = ClientTlsConfig::new()
   .ca_certificate(ca)
   .domain_name(domain_name);