xRPC-rs

Local-first async RPC and IPC library for Rust with a consistent interface across in-process, shared-memory, Unix socket, and TCP transports.

Designed primarily for communication between threads and processes on the same machine, while allowing the same RPC model to extend over TCP to trusted networks and small-scale service deployments.

Start locally with Channel, Arc, shared memory, or Unix sockets, then move the same RPC interface to TCP when services need to span multiple hosts.

Status: Actively developed. Core features are available, and the API may change between releases.

Quick Start

This single-process example uses an in-process channel pair. The same RPC layer can run across local processes over shared memory or Unix sockets, or across hosts over TCP, by replacing the frame transport. Transport configuration and deployment lifecycle differ, but application-level handlers and RPC calls keep the same model.

use serde::{Deserialize, Serialize};
use xrpc::{
    ChannelConfig, ChannelFrameTransport, MessageChannelAdapter, RpcClient, RpcServer,
};

#[derive(Serialize, Deserialize)]
struct AddRequest { a: i32, b: i32 }

#[derive(Serialize, Deserialize)]
struct AddResponse { result: i32 }

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let (client_transport, server_transport) =
        ChannelFrameTransport::create_pair("quick-start", ChannelConfig::default())?;

    let server = RpcServer::new();
    server.register_typed("add", |req: AddRequest| async move {
        Ok(AddResponse { result: req.a + req.b })
    });
    let server_handle =
        server.spawn_handler(MessageChannelAdapter::new(server_transport));

    let client = RpcClient::new(MessageChannelAdapter::new(client_transport));
    // Retain the handle when explicit task joining or forced shutdown is needed.
    let client_handle = client.try_start()?;

    let response: AddResponse = client
        .call("add", &AddRequest { a: 1, b: 2 })
        .await?;
    assert_eq!(response.result, 3);

    let close_result = client.close().await;
    let join_result = client_handle.join().await;
    close_result?;
    join_result?;
    server_handle.shutdown().await;

    Ok(())
}

Calling try_start() is required before making RPCs. Retaining its handle is optional for ordinary calls, but it enables explicit receive-task joining and forced shutdown. This example retains it for strict cleanup confirmation.

Ordinary unary calls use the client's response timeout. To wait without an application-level response deadline, use explicit call options:

use xrpc::CallOptions;

let response: AddResponse = client
    .call_with_options(
        "long_operation",
        &request,
        CallOptions::without_timeout(),
    )
    .await?;

This disables only the RPC response deadline; configured transport I/O timeout policies remain in effect. Client close, terminal connection failure, and dropping the call future still end the local wait, but they do not cancel work already running on the server.

Installation

[dependencies]
xrpc-rs = "0.3.1"
serde = { version = "1", features = ["derive"] }
tokio = { version = "1", features = ["macros", "rt-multi-thread"] }

To enable an optional codec, replace the xrpc-rs dependency above with, for example:

xrpc-rs = { version = "0.3.1", features = ["codec-messagepack"] }

Available codec features are codec-messagepack, codec-cbor, codec-postcard, and codec-all. Bincode and JSON support are available without optional features.

xrpc-rs 0.3 requires Rust 1.85 or newer.

Status

Feature	Status
FrameTransport Layer (TCP, Unix, SharedMemory, Channel, Arc)	✅ Completed
MessageChannel with compression	✅ Completed
RPC Client/Server	✅ Completed
Streaming	✅ Completed
Connection Pooling	✅ Completed
Service Discovery & Load Balancing	✅ Completed
Docs & Examples	In Progress
Python Bindings	Under Consideration

Architecture

xRPC follows a layered architecture:

Layer	Trait/Module	Description
Layer 1	`FrameTransport`	Low-level byte transmission with framing
Layer 2	`MessageChannel`	Message-aware channel with compression
Layer 3	`RpcClient/RpcServer`	RPC with method dispatch, streaming
Layer 4	`LoadBalancer`	Service discovery, load balancing

Application
    ↓
LoadBalancedClient (Layer 4)
    ↓
RpcClient/RpcServer (Layer 3)
    ↓
MessageChannel (Layer 2)
    ↓
FrameTransport (Layer 1)
    ↓
In-process / Local IPC / TCP

Transport Comparison

Transport	Typical Use	Cross-Process	Cross-Host	Serialization
`ChannelFrameTransport`	Same-process async communication / testing	No	No	Yes
`ArcFrameTransport`	Same-process typed fast path	No	No	Frame/RPC: Yes; `ZeroCopyTransport`: No
`SharedMemoryFrameTransport`	Low-latency local IPC, one client per named server	Yes	No	Yes
`UnixFrameTransport`	Local IPC on Unix systems	Yes	No	Yes
`TcpFrameTransport`	Loopback, trusted LAN, small-scale services	Yes	Yes	Yes

TCP transport does not provide TLS, authentication, or authorization. Use it within a trusted network or behind an appropriate security boundary.

Load Balancing

Distribute requests across local worker processes or small service instances:

use xrpc::{
    LoadBalancedClient, LoadBalancer, ClientFactory, Endpoint,
    StaticDiscovery, RoundRobin, RpcClient, MessageChannelAdapter,
    TcpFrameTransport, TcpConfig, BincodeCodec, RpcError,
};
use async_trait::async_trait;
use std::sync::Arc;

struct TcpFactory;

#[async_trait]
impl ClientFactory<MessageChannelAdapter<TcpFrameTransport>, BincodeCodec> for TcpFactory {
    async fn create(&self, endpoint: &Endpoint) 
        -> Result<RpcClient<MessageChannelAdapter<TcpFrameTransport>, BincodeCodec>, RpcError> 
    {
        let Endpoint::Tcp(addr) = endpoint else {
            return Err(RpcError::ClientError("Expected TCP".into()));
        };
        let transport = TcpFrameTransport::connect(*addr, TcpConfig::default())
            .await.map_err(RpcError::Transport)?;
        Ok(RpcClient::new(MessageChannelAdapter::new(transport)))
    }
}

// Setup load balancer
let endpoints = vec![
    Endpoint::tcp_from_str("127.0.0.1:8001")?,
    Endpoint::tcp_from_str("127.0.0.1:8002")?,
    Endpoint::tcp_from_str("127.0.0.1:8003")?,
];

let discovery = Arc::new(StaticDiscovery::new(endpoints));
let lb = Arc::new(LoadBalancer::new(discovery, RoundRobin::new()));
let client = LoadBalancedClient::new(lb, Arc::new(TcpFactory));
client.init().await?;

// Calls are distributed across workers or service instances.
// Setup failures can fail over before invocation.
let response: Response = client.call("method", &request).await?;
client.close().await?;

Strategies: RoundRobin, Random, LeastConnections, WeightedRoundRobin, ScoreBased

Features:

Failover for endpoint selection, connection creation, and startup failures
No automatic retry after RPC invocation, preventing duplicate non-idempotent calls
Stream affinity (streaming calls stay on same server)
Failure-based health tracking with configurable thresholds
DNS-based discovery; periodic refresh and discovery watching are started explicitly

Documentation

Reference

Message Protocol - binary framing and payload encoding
Client Lifecycle - start, close, join, and forced shutdown
Transport Shutdown - close and terminal error contract

Migration

Migrating to 0.3 - lifecycle, error matching, and wire compatibility changes

Planned:

Transport layer guide
Client/Server usage guide
Streaming guide
Discovery & load balancing guide
Codec guide (Bincode, JSON, MessagePack, CBOR, Postcard)
Architecture overview & getting started

Examples

RPC Client/Server

# Terminal 1
cargo run --example rpc_client_server -- server

# Terminal 2
cargo run --example rpc_client_server -- client

SharedMemory Transport

# Terminal 1
cargo run --example message_transport_shm -- server

# Terminal 2
cargo run --example message_transport_shm -- client

Byte-level Transports

cargo run --example byte_transports