← Wiki

Async_rpc, underneath

2026-10-17 · 11 min read · ocaml, rpc, protocol

Read against janestreet/async_rpc_kernel at master. The code below is quoted from the source with one deliberate change: every type is named t inside a module named for it. Where the source says Query.V1.needs_length, this page says Query.V1.t. The renamed types are listed in Names in the source, so they can still be grepped for. async_rpc_kernel is the transport-agnostic core, and async_rpc is the thin layer that pins it to Async_unix readers and writers.

The thesis of the whole library fits in one observation: there is no streaming in the protocol. There are three kinds of message, and Pipe_rpc is a convention layered on top of two of them.

The wire

A connection is a sequence of length-prefixed bin_prot blobs. The framing is eight bytes, little-endian, from Transport:

let length = 8
let unsafe_get_payload_length buf ~pos = Bigstring.unsafe_get_int64_le_exn buf ~pos

Each payload is a Message.t. Stripped of the compatibility variants that accumulated over eleven protocol versions, it is:

module Message = struct
  type 'a t =
    | Heartbeat
    | Query_v1 of 'a Query.V1.t
    | Response_v1 of 'a Response.V1.t
    (* ... Query_v2..v4, Response_v2, Metadata, Close_reason, Close_started *)
end

and the two that matter:

module Query = struct
  module V1 = struct
    type 'a t =
      { tag : Rpc_tag.t
      ; version : int
      ; id : Query_id.t
      ; data : 'a
      }
  end
end

module Response = struct
  module V1 = struct
    type 'a t =
      { id : Query_id.t
      ; data : 'a Rpc_result.t
      }
  end
end

That is the entire protocol. A query names an RPC by tag and version, carries a fresh id, and a response quotes that id back. Everything else — pipes, state, aborts, errors — is encoded inside data.

Why 'a is a lie

On the write side 'a is the real payload type, and the payload is written with its length in front of it (Writer_with_length). On the read side the connection instantiates 'a at Nat0.t: what it parses is a Nat0.t Message.t.

So the reader parses the envelope — tag, version, id — and then reads only the payload's length, leaving the bytes in the buffer. The payload reaches the handler undigested:

module Response_handler = struct
  type t =
    data:Nat0.t Rpc_result.t
    -> read_buffer:Bigstring.t
    -> read_buffer_pos_ref:int ref
    -> Response_handler_action.t
end

This is what lets one monomorphic connection carry arbitrarily many RPCs with unrelated payload types. The connection never knows what an 'a is; the dispatcher that registered the handler does, and it deserialises from the shared read buffer at the recorded position. No intermediate copy, and no existential wrapper around every message.

Handshake

Each side opens by writing a Protocol_version_header.t — a magic prefix plus the list of protocol versions it supports — and negotiation picks the greatest common one:

let earliest_valid_version_number = 1
let actual_latest_version_number = 11
let latest = create ~supported_versions:(versions_range 1 latest_version_number)

The negotiated number then gates features (version_dependent_feature.ml): whether the peer understands Close_reason, whether metadata may ride along, whether queries may address an RPC by menu rank. A peer stuck at v1 still works; it just gets the v1 message shapes. The handshake has its own timeout, separate from heartbeats:

let default_handshake_timeout = Time_ns.Span.of_int_sec 30

A plain Rpc

Dispatch allocates an id, registers a handler, and writes one Query:

; open_queries : (P.Query_id.t, Description.t * (Response_handler.t[@sexp.opaque])) Hashtbl.t

When a Response arrives, handle_response is the whole routing layer:

match Hashtbl.find t.open_queries id with
| None -> Stop (Error (Rpc_error.Unknown_query_id id))
| Some (rpc, response_handler) -> ...

The handler fills an ivar and asks to be removed. One query, one response, one table entry, gone.

Streaming is Keep

Here is the hinge. A response handler does not return unit; it returns an instruction to the connection:

module Response_handler_action = struct
  module Response_with_determinable_status = struct
    type t =
      | Pipe_eof
      | Expert_indeterminate
      | Determinable :
          'a Rpc_result.t * 'a Implementation_mode.Error_mode.t
          -> t
  end

  type t =
    | Keep
    | Wait of unit Deferred.t
    | Remove of
        (Response_with_determinable_status.t, Rpc_error.t Modes.Global.t)
          Result.t
    | Expert_remove_and_wait of unit Deferred.t
end

A streaming RPC's handler returns Keep. The entry stays in open_queries, so the next Response bearing that id is routed to the same handler, and the one after that. Streaming is one query whose response handler declines to retire.

The payloads carry the structure the envelope lacks. The query is not the user's query type but a variant:

module Stream_query = struct
  type 'a t =
    [ `Query of 'a
    | `Abort
    ]
end

the first response is an initial message:

module Stream_initial_message = struct
  type ('response, 'error) t =
    { unused_query_id : Unused_query_id.t
    ; initial : ('response, 'error) Result.t
    }
end

and every response after it is:

module Stream_response_data = struct
  type 'a t =
    [ `Ok of 'a
    | `Eof
    ]
end

So the life of a pipe, in messages:

client -> Query   { id; tag; data = `Query q }
server -> Response { id; data = Ok initial_message }    (* Ok () or Error 'error *)
server -> Response { id; data = Ok `Ok elt }            (* ... many times *)
server -> Response { id; data = Ok `Eof }               (* handler finally Removed *)

unused_query_id is exactly what it sounds like: a vestigial field the initial message still serialises because removing it would change the bin digest. The library is twenty years old in places and honest about it.

Pipe_rpc and State_rpc are one function

This is the part worth reading the source for. Both are Streaming_rpc.t:

module Streaming_rpc = struct
  type ('query, 'initial_response, 'update_response, 'error_response) t =
    { tag : P.Rpc_tag.t
    ; version : int
    ; bin_query : 'query Bin_prot.Type_class.t
    ; bin_initial_response : 'initial_response Bin_prot.Type_class.t
    ; bin_update_response : 'update_response Bin_prot.Type_class.t
    ; bin_error_response : 'error_response Bin_prot.Type_class.t
    ; client_pushes_back : bool
    (* ... four Type_equal.Id.t for tracing *)
    }
end

and the two public modules differ only in how they fill it in:

module Pipe_rpc = struct
  type ('query, 'response, 'error) t = ('query, unit, 'response, 'error) Streaming_rpc.t

  let create ?client_pushes_back ~name ~version ~bin_query ~bin_response ~bin_error () =
    Streaming_rpc.create
      ?client_pushes_back ~name ~version ~bin_query
      ~bin_initial_response:Unit.bin_t          (* <-- *)
      ~bin_update_response:bin_response
      ~bin_error
      ~alias_for_initial_response:""
      ~alias_for_update_response:"response"
      ()
end

module State_rpc = struct
  type ('query, 'state, 'update, 'error) t =
    ('query, 'state, 'update, 'error) Streaming_rpc.t

  let create ... =
    Streaming_rpc.create
      ...
      ~bin_initial_response:bin_state           (* <-- *)
      ~bin_update_response:bin_update
      ~alias_for_initial_response:"state"
      ~alias_for_update_response:"update"
      ()
end

A Pipe_rpc is a State_rpc whose state is unit. The initial message already existed for the error channel, so carrying a state in it was free; the aliases exist only to name the Type_equal.Id.ts that tracing prints. On the wire a State_rpc and a Pipe_rpc are indistinguishable apart from how many bytes the first response holds.

This also explains the dispatch return type, which otherwise looks over-engineered — two nested results:

('response Pipe.Reader.t * Metadata.t, 'error) Result.t Or_error.t

The outer Or_error.t is the RPC failing: no such implementation, connection closed, deserialisation error. The inner ('a, 'error) Result.t is the implementation declining, in the application's own error type, which is the initial field of the initial message. A pipe RPC can refuse before it has produced a single element, and that refusal is typed.

Abort is a second query with the same id

There is no cancel message. The client's pipe reader being closed is the trigger, wired up at dispatch:

upon (Pipe.closed pipe_r) (fun () ->
  if not (Ivar.is_full close_reason)
  then (
    abort t conn id;
    Ivar.fill_if_empty close_reason Closed_locally));

and abort is not a new message kind, it is the original query again:

let abort t conn id =
  let query =
    { P.Query.Validated.tag = t.tag
    ; version = t.version
    ; id                              (* the same id as the original query *)
    ; metadata = Null
    ; data = `Abort
    }
  in
  ignore (Connection.dispatch conn ~kind:Abort_streaming_rpc_query
            ~bin_writer_query:P.Stream_query.bin_writer_nat0_t
            ~query ~response_handler:None : _ Result.t)

Note ~response_handler:None — the abort is fire-and-forget at the protocol level. The server's streaming implementation recognises the id, closes its pipe, and the ordinary ` Eof `` follows, which is what actually retires the client's handler.

The source is candid about the race this creates. If the server has already closed the pipe, it will not answer the abort, because the query id is no longer open on its side:

once the server closes the pipe, it will no longer respond to abort messages with [Eof]`. We need to immediately remove the handler instead of sending abort in that case, to avoid leaking a response handler.

So the client's bookkeeping is: if we have seen ` Eof `, drop the handler locally; otherwise send the abort and wait for the Eof ` it will provoke. Getting that wrong leaks an entry in open_queries` for the life of the connection.

Backpressure

Two mechanisms, at different layers.

The reader stops reading. Every handler result can be Wait of unit Deferred.t, and the transport's on_message callback returns the same shape:

module Handler_result = struct
  type 'a t =
    | Stop of 'a
    | Continue
    | Wait of unit Deferred.t
end

A Wait suspends the read loop. Nothing is consumed from the socket, the kernel buffer fills, the window closes, and the writer at the far end blocks in the ordinary TCP way. Application-level flow control collapses into transport-level flow control, with no credit scheme to get wrong.

client_pushes_back decides whether a given RPC opts into that. It is a field on the Streaming_rpc.t, fixed at creation, and the rule it selects is exactly this:

| Update data ->
  if not (Pipe.is_closed pipe_w)
  then (
    Pipe.write_without_pushback pipe_w data;
    if t.client_pushes_back && Pipe.length pipe_w >= Pipe.size_budget pipe_w
    then Wait (get_flush_deferred ())
    else Continue)
  else Continue

Worth reading twice. The element is written unconditionally and without pushback — the pipe never refuses an element mid-stream. Only afterwards, if the consumer has fallen behind by more than the pipe's size budget, does the handler return Wait, which stalls the read loop until the pipe flushes. Pushback is therefore a high-water mark on an already-written queue, not a per-element gate, and with client_pushes_back = false a slow consumer simply grows the pipe without bound.

On the writing side, Direct_stream_writer exists to skip the Pipe.t entirely: an implementation that produces elements eagerly can write them straight into the transport rather than allocating a pipe and a transfer loop per connection.

Heartbeats and closing

?(timeout = Time_ns.Span.of_sec 30.)
?(send_every = Time_ns.Span.of_sec 10.)

A Heartbeat is sent every ten seconds and the connection dies if nothing arrives for thirty. "Nothing" is literal — liveness is not heartbeat-specific:

~on_end_of_batch:(fun () ->
  Atomic.set t.last_seen_alive (Synchronous_time_source.now t.time_source);
  t.sent_heartbeats_without_receiving_any_messages <- 0)

Any inbound batch refreshes it, so a busy connection never depends on heartbeats at all; they exist to keep an idle one observably alive. The two timeout styles visible in that record — elapsed time since last_seen_alive, versus a count of heartbeats sent without hearing anything back — are the old and new ways of judging the same thing.

There is an escape hatch for debugging, ASYNC_RPC_HEARTBEAT_TIMEOUT_OVERRIDE, read from the environment, which exists because a breakpoint otherwise kills every connection in the process.

Shutdown is also a message: Close_started then Close_reason, so the peer can log why the connection went away instead of reporting a bare EOF. That is a small thing that makes production debugging dramatically less miserable.

Ranks: addressing an RPC by index

The newest query shape stops sending the name:

module Rpc_specifier = struct
  type t =
    | Tag_and_version of Rpc_tag.t * int
    | Rank of int
end

Once both sides have exchanged a menu of implemented RPCs, a query can name one by its position in that menu rather than by a string and an integer. For a high-rate RPC with a long name, the tag was a meaningful fraction of every message.

Compared with the Python port

libs/async-rpc in ohayou.ai makes the opposite trade. Its Frame is:

type Frame[Req, Val, Err, Item] = Annotated[
    Request[Req] | Response[Val, Err] | StreamItem[Item] | StreamClose,
    Field(discriminator="kind"),
]

Streaming is explicit in the envelope — StreamItem and StreamClose are message kinds, where Async_rpc hides both inside a Response's payload. The consequences are worth being deliberate about:

Async_rpcthe Python port
message kinds3 (+ metadata/close)4
streaming visible on the wireno, it is payload conventionyes, as frame kinds
receiver-initiated cancela second query, same id, ` Abort ``none
stream identitythe originating query ida separate StreamId
payload parsed bythe dispatcher, from a shared bufferpydantic, eagerly, per frame

Async_rpc's shape is a consequence of bin_prot and of wanting one monomorphic connection: fewer variants means a stabler digest and a dispatch table with one shape of entry. The Python version is reading a self-describing format into validated objects anyway, so making the stream kinds explicit costs nothing and reads better in a packet dump.

One row of that table is a real gap rather than a trade. StreamClose is sender-initiated — "no further items on this stream" — and the only abort in the package is self._writer.transport.abort(), which kills the whole connection. A subscriber that wants to stop receiving a stream has no way to say so, and the producer keeps producing. Async_rpc's answer costs one variant in a payload that already exists.

The other idea worth stealing wholesale is the handler that returns Keep. It is what makes a streaming RPC cost exactly one table entry and no extra protocol surface, and it is why State_rpc was almost free to add once Pipe_rpc existed.

Names in the source

Everything on this page is spelled in the one-type-per-module style. The source is not, in places, because the wire types come in a write form that needs a length and a read form that does not. To grep for one, here is what each renamed type is called there:

One value keeps its source name in the quotes above: Stream_query.bin_writer_nat0_t is the writer for Nat0.t Stream_query.t, named for the nat0_t type it was derived from.

Types that were already t in the source (Streaming_rpc.t, Pipe_rpc.t, Response_handler.t, Rpc_specifier.t, Handler_result.t) are unchanged, as are the builtins.