Use request/reply

An RPC is a message that returns a value. For a guided introduction see adding request/reply; this page is the recipes.

Define an RPC

import {
  defineContract,
  defineExchange,
  defineMessage,
  defineQueue,
  defineQueueBinding,
  defineRpc,
} from "@amqp-contract/contract";
import { z } from "zod";

const rpcDlx = defineExchange("rpc-dlx");
// A dead-letter exchange with nothing bound to it discards every rejected
// request, so `defineContract` rejects it. `rpc-dlx` is topic, so `#` catches
// whatever routing key the request arrived with.
const rpcDlq = defineQueue("rpc-dlq");

const calculate = defineRpc(defineQueue("rpc.calculate", { deadLetter: { exchange: rpcDlx } }), {
  request: defineMessage(z.object({ a: z.number(), b: z.number() })),
  response: defineMessage(z.object({ sum: z.number() })),
});

export const contract = defineContract({
  rpcs: { calculate },
  queues: { rpcDlq },
  bindings: { rpcDlqBinding: defineQueueBinding(rpcDlq, rpcDlx, { routingKey: "#" }) },
});

An RPC owns its queue and goes in rpcs, not in publishers or consumers — it is both.

Handle it

handlers: {
  calculate: ({ input: { payload } }) => OkAsync({ sum: payload.a + payload.b }),
},

The returned value is the reply, typed by the response schema. Async work looks the same as in a consumer:

calculate: ({ input: { payload } }) =>
  fromPromise(lookupRate(payload), qualifyRetryable("rate service down")).map((rate) => ({
    sum: payload.a * rate,
  })),

Call it

import { P } from "unthrown";

const result = await client.call("calculate", { a: 2, b: 3 }, { timeoutMs: 5_000 });

result.match({
  ok: (reply) => console.log(reply.sum),
  errCases: (matcher) =>
    matcher
      .with(P.tag("@amqp-contract/RpcTimeoutError"), () => console.error("no reply in time"))
      .with(P.tag("@amqp-contract/RpcCancelledError"), () => console.error("client closing"))
      .with(P.tag("@amqp-contract/PublishError"), (e) =>
        console.error(`request not sent (${e.reason})`),
      )
      .with(P.tag("@amqp-contract/MessageValidationError"), (e) =>
        console.error("reply failed its schema", e.issues),
      ),
  defect: (cause) => {
    throw cause;
  },
});

Always set timeoutMs. A call with no reply is otherwise bounded only by the server-side default, and a caller holding a request open is holding memory.

The request is published with expiration set to timeoutMs (rounded up to whole milliseconds), so a request no worker picked up before the caller gave up is never answered for nobody: the broker expires it out of the RPC queue. Every consumed queue has a dead-letter exchange (unless it is onPoison: "drop", where the request is discarded), so an expired request lands in the dead-letter queue with an x-death reason of expired. Pass publishOptions: { expiration } to override it.

Do not replay those dead letters. The caller already got RpcTimeoutError and its reply consumer is gone, so a replayed request runs the handler — side effects included — for nobody. A dead-letter consumer or replay tool that shares the DLQ with other messages should skip requests whose x-death reason is expired (or that carry a replyTo); see replay a dead-lettered message.

The round trip is recorded on its own histogram, amqp.client.rpc.duration — not on the publish histogram — so a slow handler does not read as a slow broker.

Declare typed errors

Business failures belong in the contract, not squeezed into the response schema:

const getOrder = defineRpc(defineQueue("rpc.get-order", { deadLetter: { exchange: rpcDlx } }), {
  request: defineMessage(z.object({ orderId: z.string() })),
  response: defineMessage(z.object({ orderId: z.string(), status: z.string() })),
  errors: {
    ORDER_NOT_FOUND: { data: z.object({ orderId: z.string() }) },
  },
});

Return one from the handler:

import { rpcError } from "@amqp-contract/worker";

getOrder: ({ input: { payload } }) => {
  const order = orders.get(payload.orderId);
  return order
    ? OkAsync({ orderId: order.id, status: order.status })
    : ErrAsync(rpcError("ORDER_NOT_FOUND", { orderId: payload.orderId }));
},

Or, with autocomplete over the declared codes, via helpers.errors:

getOrder: ({ errors, input: { payload } }) =>
  ErrAsync(errors.ORDER_NOT_FOUND({ orderId: payload.orderId })),

Handle it on the caller:

import { isRpcError } from "@amqp-contract/client";

if (result.isErr() && isRpcError(result.error)) {
  // result.error.code is "ORDER_NOT_FOUND"
  // result.error.data is { orderId: string }
}

A declared error is a business outcome, not a processing failure: the worker validates the data, publishes the error reply, and acknowledges the request. Declared errors are never retried. Neither is an RPC request whose handler fails with RetryableError: the caller is waiting on a timeoutMs shorter than most backoffs, so the worker dead-letters the request instead of re-running the handler for nobody — whatever the queue's retry config.

Set a per-call timeout

await client.call("calculate", { a: 1, b: 2 }, { timeoutMs: 30_000 });

Size it to the work, not to a house default. When it expires the pending call is cleared and you get RpcTimeoutError — but note the request may still be processed by the server: the request's expiration only dead-letters it if it is still queued. A timeout tells you no reply arrived, not that nothing happened.

Retry a timed-out call

Use a call interceptor so the policy lives in one place rather than at every call site — see add middleware.

Be careful: retrying a call whose handler is not idempotent runs the work twice.

Understand what makes a call fail

Failure	Cause
RpcTimeoutError	No reply within timeoutMs. Also what you get when the reply was dropped for being malformed.
RpcCancelledError	The client closed while the call was in flight.
RpcError<code, data>	A declared business error from the handler.
MessageValidationError	The request failed its schema, or the reply arrived but failed the response schema.
PublishError	The request never reached the broker: publish timeout, broker nack, channel closed.
Defect	An unexpected transport fault, or an error reply whose code the local contract does not declare.

That first row is worth dwelling on. If the handler returns a value that fails the response schema, the worker refuses to publish a malformed reply — so the caller sees a timeout rather than a wrong answer. A call timing out while the server looks healthy usually means a response-schema mismatch between the two sides' contracts.

Know the runtime guarantees

The type system stops undeclared error codes, but a cast or a version skew can get past it. The runtime holds the line:

• Worker — an undeclared code, or error data failing its schema, is a contract violation. No reply is published and the request is dead-lettered as a NonRetryableError. The caller times out.
• Client — an error reply whose code is not in the local contract resolves to a defect. Error data failing its schema resolves to Err(MessageValidationError).

An RPC also requires replyTo and correlationId on the request. A request missing either is dead-lettered rather than answered, since there is nowhere to reply to.

The worker only replies to direct reply-to addresses (amq.rabbitmq.reply-to…, which is what client.call() sets), so a request cannot make it publish into an arbitrary queue. A request with any other replyTo is dead-lettered with the reason logged. To reply to your own reply queues, allow them explicitly:

TypedAmqpWorker.create({
  contract,
  handlers,
  urls,
  rpc: { allowReplyTo: (replyTo) => replyTo.startsWith("replies.") },
});

Error data is validated twice — on the worker before publishing, on the client on arrival — the same as responses.

Where next

• Adding request/reply — the guided version.
• Error model — the RPC wire format and RPC_ERROR_CODE_HEADER.
• Add middleware — caching an RPC by short-circuiting.