Messaging

The messenger sends active messages. An active message names a handler on a remote instance. That handler runs when the message arrives.

Four patterns

The patterns differ in what the caller waits for.

All four builders work the same way. Get a builder from the Velo instance, attach a payload, name the destination instance, then call .send().await.

// Fire-and-forget. Delivery is best effort.
node_a.am_send("notify")?.payload(&event)?.instance(b).send().await?;

// Sync. Returns when the handler finishes.
node_a.am_sync("process")?.payload(&job)?.instance(b).send().await?;

// Unary with raw bytes.
let reply: Bytes = node_a.unary("ping")?.raw_payload(Bytes::new()).instance(b).send().await?;

// Typed unary. Payloads are JSON (serde_json).
let reply: MyResponse = node_a.typed_unary::<MyResponse>("rpc")?.payload(&req)?.instance(b).send().await?;

A payload that is larger than the large-payload threshold does not travel in the message. The messenger stages it with rendezvous and sends a handle. The receiver pulls the bytes before the handler runs.

Timeouts

velo puts no deadline on a request. A sync or unary call waits until the remote answers or the send fails. To bound the wait, set .timeout(d) on an am_sync, unary, or typed_unary builder.

use std::time::Duration;
use velo::RequestTimeout;

match node_a.unary("ping")?.timeout(Duration::from_secs(2)).instance(b).send().await {
    Ok(reply) => { /* the answer */ }
    Err(e) if e.is::<RequestTimeout>() => { /* no answer in time */ }
    Err(e) => return Err(e),
}

The clock starts at send(). It covers the wait for a response slot, discovery, the handshake, the send, and the answer. When the time runs out, the call fails with RequestTimeout. Every other error is a failed send or an error from the remote. Only awaiting the result is bounded, not admitted().

At expiry, the call gives back its response slot, also when the caller keeps the result. A reply that arrives later is discarded. The remote can still run the handler. A call without .timeout makes no timer.

When the request's payload is staged, the timeout also bounds the receiver's fetch of it. At expiry the receiver gives up the fetch and its response slot, the handler does not run, and the sender frees the payload. See Transparent staging.

A remote that forgot the caller never answers. The remote sends a reply through its own peer table, not back over the connection of the request. So a call to a remote that called unregister_peer on the caller waits until its timeout, and without a timeout it waits for ever.

Handlers

You register a handler by name. Each handler kind has a sync form and an async form.

use velo::{Context, Handler, TypedContext};

// Sync unary handler that returns raw bytes.
node.register_handler(Handler::unary_handler("ping", |_ctx: Context| Ok(Some(Bytes::from("pong")))).build())?;

// Async typed handler. The input is deserialized and the output is serialized.
node.register_handler(
    Handler::typed_unary_async("add", |ctx: TypedContext<AddRequest>| async move {
        Ok(AddResponse { sum: ctx.input.a + ctx.input.b })
    })
    .build(),
)?;

// Async fire-and-forget handler.
node.register_handler(Handler::am_handler_async("notify", |ctx: Context| async move { Ok(()) }).build())?;

The handler context holds the payload, the message headers, and the messenger (ctx.msg). A handler can use the messenger to send messages, register handlers, or wait for events. ctx.sender_worker_id() is always available. ctx.sender_instance_id() returns the InstanceId of the sender after its handshake is complete.

Handler names that start with _ are reserved for system handlers.

Dispatch modes

.max_concurrent(n) limits how many lanes run the handler at the same time. A lane takes the permit for each message inside the lane, so the limit does not change the order in a lane. The builder rejects 0 because it would stop every lane.

let h = Handler::typed_unary_async("bulk", bulk)
    .ordered_with(
        OrderedConfig::by_sender()
            .with_idle_lane_ttl(Some(Duration::from_secs(300)))
            .with_max_queue_depth(Some(100_000))
            .with_overflow(OverflowPolicy::Reject),
    )
    .build();

Ordered dispatch has these limits:

• It keeps order. It does not create order. If a peer uses more than one transport, or a connection drops and reconnects, the arrival order was already lost before the messenger saw the messages.
• Large payloads are not ordered. A rendezvous-staged message resolves before dispatch. The dispatcher logs a warning once when it sees one.
• Lane queues have no hard bound. max_queue_depth is a soft limit for each lane. It drives OverflowPolicy. It does not bound the channel. One slow peer cannot cause the messages of other peers to be shed.
• A unary handler serves one request at a time for each sender. A client that makes 100 concurrent calls gets them one after the other.
• Idle lanes stop. A lane with no messages for idle_lane_ttl (30 s by default) stops, so short-lived peers do not leak tasks.

Watch velo_messenger_ordered_lane_depth and velo_messenger_ordered_lane_wait_seconds to find a lane that falls behind. Under OverflowPolicy::Reject, velo_messenger_dispatch_failures_total{reason="ordered_lane_shed"} counts the shed messages.