52 — LLD: Mini RPC Framework

Understanding the Problem

A Remote Procedure Call (RPC) framework lets one program invoke a method on another program — across processes or machines — as if it were a local function call. The framework hides the wire format, the transport, error mapping, and timeouts. The interesting design at LLD scope is the registry — how the server learns which method names belong to which service, and how the client's local stub finds the right registered handler at call time.

What is RPC?

RPC is the older, simpler cousin of REST and gRPC. The contract is verb-shaped: userService.getUser(id) instead of GET /users/:id. The framework's job is to translate a method-name + args tuple into a transported message and back into a function call on the server side. At LLD interview scope the transport is usually in-memory or a stub — the design conversation is about the registry, dispatch, and serialization seams.

---

Requirements

Clarifying Questions

You: Is this in-memory dispatch only, or do we need actual networking?

Interviewer: In-memory for v1. Design the seam so a network transport drops in later.

Keeps Transport as an interface. The reference implementation is a same-process call; a Netty/HTTP transport is just another class.

You: Are services keyed by name? Or by class type?

Interviewer: By name. Multiple instances of the same service can be hosted with different serviceName keys.

Registry key is (serviceName, methodName). Allows two OrderService instances to coexist as "orders.us" and "orders.eu".

You: Sync or async?

Interviewer: Both — but the public API is a future. Sync becomes a .get() with a timeout.

Returns CompletableFuture<Response>. Sync wrappers are syntactic sugar.

You: What about errors — do we surface server exceptions to the caller?

Interviewer: Yes. Map them to a small set of RpcError codes: METHOD_NOT_FOUND, BAD_ARGS, SERVER_ERROR, TIMEOUT.

The error type becomes part of the wire contract.

You: Streaming or single-response?

Interviewer: Single-response only. Streaming is a follow-up.

Keeps the interface as request → response. Mention streaming as an extensibility point.

You: Marshalling format?

Interviewer: Pluggable — JSON for v1, but Protobuf later.

Codec interface so the registry never sees raw bytes-vs-objects choices.

Final Requirements

1. register(serviceName, instance) discovers public methods on a service object and stores them keyed by (serviceName, methodName).
2. call(serviceName, methodName, args) returns a CompletableFuture<Response>.
3. Pluggable Codec (JSON / Protobuf) and Transport (in-memory / network) — both behind interfaces.
4. Error mapping: METHOD_NOT_FOUND, BAD_ARGS, SERVER_ERROR, TIMEOUT.
5. Call timeouts.
6. Thread-safe.

Out of scope: streaming, auth, retries, service discovery, load balancing.

Deferred tradeoffs: reflection-based vs. code-generated stubs (we pick reflection for the interview; codegen is the production answer for type safety and hot-path perf).

---

Core Entities and Relationships

Walking the nouns: service, method, request, response, codec, transport, registry. Quick triage:

• Service — not a class; we hold the registered Java/TS object directly.
• Method — wraps (instance, java.lang.reflect.Method) so we can invoke later. Worth a small struct.
• Request / Response — wire envelopes with id, service, method, payload. Definitely classes.
• Codec — interface; serializes/deserializes payloads.
• Transport — interface; ships a request, returns a response future.
• MethodRegistry — lookup keyed on (serviceName, methodName).
• RpcServer — owns the registry; its handle(request) is what a transport calls.
• RpcClient — holds a transport + codec; exposes call(...).

Entity	Responsibility
RpcRequest	{id, serviceName, methodName, argsBytes}.
RpcResponse	{id, resultBytes? , error?}.
RpcError	{code, message}.
MethodHandle	{instance, method, paramTypes} — invocable.
MethodRegistry	register(name, instance), lookup(service, method).
Codec	encode(obj) -> bytes, decode(bytes, type) -> obj.
Transport	send(request) -> Future<Response>.
RpcServer	Holds registry; handle(request) reflects → invokes → wraps response.
RpcClient	Holds transport + codec; call(service, method, args) returns future.

Why MethodHandle and not java.lang.reflect.Method directly? Because the registry also caches param types for cheap arg-decoding, and because we may later swap to MethodHandles or generated stubs without changing callers.

Design is iterative — start with sync in-memory, then bolt on the transport seam.

---

Class Design

RpcRequest

State: id: String, serviceName: String, methodName: String, payload: byte[] (codec-encoded args)

RpcResponse

State: id: String, result: byte[] (nullable), error: RpcError (nullable)

Exactly one of result / error is set. The id matches the request — the client uses it to resolve the right future when the transport multiplexes calls.

MethodRegistry

State: Map<String, MethodHandle> handlers — key is serviceName + "." + methodName.

Methods:

• register(serviceName, instance) — reflects all public methods, caches them.
• lookup(serviceName, methodName) -> MethodHandle?.

RpcServer

State: registry, codec.

Methods:

• register(serviceName, instance) — delegates to the registry.
• handle(request) -> response — codec-decodes args, invokes via reflection, codec-encodes the result. Maps exceptions to RpcError.

RpcClient

State: transport, codec, defaultTimeoutMs.

Methods:

• call(serviceName, methodName, args, returnType, timeoutMs) -> Future<T> — encodes args, calls transport, decodes result.

Final Class Design

class RpcRequest  { String id, serviceName, methodName; byte[] payload; }
class RpcResponse { String id; byte[] result; RpcError error; }
class RpcError    { String code, message; }

interface Codec       { byte[] encode(Object o); <T> T decode(byte[] b, Class<T> t); }
interface Transport   { CompletableFuture<RpcResponse> send(RpcRequest req); }

class MethodHandle    { Object instance; Method method; Class<?>[] paramTypes; Class<?> returnType; }
class MethodRegistry  { void register(String svc, Object inst); MethodHandle lookup(String svc, String m); }

class RpcServer       { void register(String svc, Object inst); RpcResponse handle(RpcRequest req); }
class RpcClient       { <T> CompletableFuture<T> call(String svc, String m, Object[] args, Class<T> rt, long toMs); }

Design principles at play:

• Strategy — Codec and Transport are swappable.
• Registry / Service Locator — MethodRegistry is exactly that.
• Adapter — RpcServer.handle adapts the wire-shaped RpcRequest to a Java method call.
• Open/Closed — adding gRPC transport is a new class implementing Transport. Zero edits in RpcClient or RpcServer.

---

Implementation

Core Logic: Server handle(request)

Steps:

1. Lookup the MethodHandle for (serviceName, methodName). If missing → RpcError("METHOD_NOT_FOUND").
2. Decode args: walk paramTypes, decode each from the payload.
3. method.invoke(instance, args).
4. Encode the return value into RpcResponse.result.
5. On InvocationTargetException, unwrap, map to RpcError("SERVER_ERROR", cause.message).
6. On any other reflection failure → RpcError("BAD_ARGS").

Edge cases: void methods return null; null payloads on registration of zero-arg methods; method overloads (decision: reject — registry uses method name only, so overloading is ambiguous).

Core Logic: Client call(...)

Steps:

1. Build RpcRequest with a fresh id (UUID).
2. Codec-encode args into payload.
3. transport.send(request) → Future<RpcResponse>.
4. Apply timeout: wrap with orTimeout(toMs, TIMEOUT).
5. On response: if error != null → fail the future. Else codec-decode result into returnType.

Bad / Good / Great — handling concurrent calls on the same transport

Bad approach: transport is single-blocking — client blocks per call, no pipelining.

• Throughput tied to RTT. Wastes the connection.

Good approach: transport multiplexes. A Map<requestId, CompletableFuture> tracks in-flight calls. Inbox thread reads responses, looks up the future, completes it.

• Pipelined throughput. Bounded inflight map prevents unbounded growth.

Great approach: same, plus backpressure: the client's call blocks (or rejects) when inflight exceeds a configured limit, preventing OOM under server slowness. Cancellation: when a future is cancelled or times out, remove from the inflight map and send a CANCEL frame so the server can stop wasting CPU on a doomed call.

Verification

Trace a register + call flow.

1. Server boots. server.register("orders", new OrderServiceImpl()). Registry now has "orders.placeOrder" → MethodHandle{...}, "orders.cancelOrder" → ....
2. Client calls client.call("orders", "placeOrder", new Object[]{userId, items}, Order.class, 1000).
3. Client encodes args via JSON codec → payload = bytes. Builds RpcRequest{id="r-1", "orders", "placeOrder", payload}. Transport sends.
4. Server receives. handle(request) → registry lookup hits → decode args → reflective invoke returns an Order → encode → RpcResponse{id="r-1", result=bytes, error=null}.
5. Client transport receives → looks up r-1 in inflight map → decodes result to Order → completes future.
6. Now call client.call("orders", "ghostMethod", ..., String.class, 1000). Server lookup misses → returns RpcResponse{id="r-2", error={"METHOD_NOT_FOUND", ...}}. Client future fails with RpcException("METHOD_NOT_FOUND").
7. Server handler throws InsufficientFundsException. Server maps to RpcError("SERVER_ERROR", "InsufficientFunds: ..."). Client receives, fails the future. Caller's catch block sees the original message preserved.

Thread Safety

• MethodRegistry is read-mostly: a ConcurrentHashMap for handlers; register writes, lookup reads concurrently.
• RpcServer.handle is stateless except for the registry; safe by construction.
• RpcClient inflight map is a ConcurrentHashMap<String, CompletableFuture<RpcResponse>>. The transport's reader thread is the only completer; the caller thread is the only registrant. No external synchronization needed.
• Reflection itself is thread-safe; the underlying user service must be thread-safe — that is the user's contract, not the framework's.

Complete Code Implementation

Java

import java.lang.reflect.*;
import java.util.*;
import java.util.concurrent.*;

class RpcRequest  { public String id, serviceName, methodName; public byte[] payload; }
class RpcResponse { public String id; public byte[] result; public RpcError error; }
class RpcError    {
    public String code, message;
    public RpcError(String c, String m) { code = c; message = m; }
}

interface Codec     { byte[] encode(Object o); <T> T decode(byte[] b, Class<T> t); }
interface Transport { CompletableFuture<RpcResponse> send(RpcRequest req); }

class MethodHandleEntry {
    final Object instance; final Method method; final Class<?>[] paramTypes;
    MethodHandleEntry(Object i, Method m) { instance = i; method = m; paramTypes = m.getParameterTypes(); }
}

class MethodRegistry {
    private final Map<String, MethodHandleEntry> handlers = new ConcurrentHashMap<>();

    public void register(String serviceName, Object instance) {
        for (Method m : instance.getClass().getMethods()) {
            if (m.getDeclaringClass() == Object.class) continue;
            String key = serviceName + "." + m.getName();
            if (handlers.putIfAbsent(key, new MethodHandleEntry(instance, m)) != null)
                throw new IllegalStateException("duplicate method: " + key);
        }
    }
    public MethodHandleEntry lookup(String svc, String name) { return handlers.get(svc + "." + name); }
}

class RpcServer {
    private final MethodRegistry registry = new MethodRegistry();
    private final Codec codec;
    public RpcServer(Codec c) { this.codec = c; }
    public void register(String svc, Object inst) { registry.register(svc, inst); }

    public RpcResponse handle(RpcRequest req) {
        RpcResponse res = new RpcResponse(); res.id = req.id;
        MethodHandleEntry h = registry.lookup(req.serviceName, req.methodName);
        if (h == null) { res.error = new RpcError("METHOD_NOT_FOUND", req.serviceName + "." + req.methodName); return res; }
        try {
            // payload is encoded as Object[] of args; codec decodes to Object[] then we cast per paramType
            Object[] args = codec.decode(req.payload, Object[].class);
            Object out = h.method.invoke(h.instance, args);
            res.result = codec.encode(out);
        } catch (InvocationTargetException ite) {
            res.error = new RpcError("SERVER_ERROR", String.valueOf(ite.getCause()));
        } catch (IllegalArgumentException iae) {
            res.error = new RpcError("BAD_ARGS", iae.getMessage());
        } catch (Exception e) {
            res.error = new RpcError("SERVER_ERROR", e.getMessage());
        }
        return res;
    }
}

class RpcClient {
    private final Transport transport; private final Codec codec; private final long defaultTimeoutMs;
    public RpcClient(Transport t, Codec c, long toMs) { transport = t; codec = c; defaultTimeoutMs = toMs; }

    public <T> CompletableFuture<T> call(String svc, String method, Object[] args, Class<T> returnType) {
        RpcRequest req = new RpcRequest();
        req.id = UUID.randomUUID().toString();
        req.serviceName = svc; req.methodName = method;
        req.payload = codec.encode(args);
        return transport.send(req)
            .orTimeout(defaultTimeoutMs, TimeUnit.MILLISECONDS)
            .thenApply(res -> {
                if (res.error != null) throw new RpcException(res.error);
                return codec.decode(res.result, returnType);
            })
            .exceptionallyCompose(ex -> {
                if (ex instanceof TimeoutException) return CompletableFuture.failedFuture(new RpcException(new RpcError("TIMEOUT", "timed out")));
                return CompletableFuture.failedFuture(ex);
            });
    }
}

class RpcException extends RuntimeException { public final RpcError error; public RpcException(RpcError e) { super(e.code + ": " + e.message); this.error = e; } }

// In-memory transport that loops back through a server
class InMemoryTransport implements Transport {
    private final RpcServer server;
    private final ExecutorService pool = Executors.newFixedThreadPool(8);
    public InMemoryTransport(RpcServer s) { this.server = s; }
    public CompletableFuture<RpcResponse> send(RpcRequest req) {
        return CompletableFuture.supplyAsync(() -> server.handle(req), pool);
    }
}

C++

#include <any>
#include <chrono>
#include <functional>
#include <future>
#include <memory>
#include <stdexcept>
#include <string>
#include <unordered_map>

struct RpcError    { std::string code, message; };
struct RpcRequest  { std::string id, serviceName, methodName; std::string payload; };
struct RpcResponse { std::string id; std::string result; RpcError error; bool hasError = false; };

// Codec is type-erased for sketch; production would template on payload type or use Protobuf.
struct Codec {
    virtual std::string encode(const std::any& o) = 0;
    virtual std::any decode(const std::string& b, const std::type_info& t) = 0;
    virtual ~Codec() = default;
};

// Method handler is a type-erased callable: (encoded args) -> encoded result.
using Handler = std::function<std::string(const std::string&)>;

class MethodRegistry {
public:
    void registerMethod(const std::string& svc, const std::string& name, Handler h) {
        handlers[svc + "." + name] = std::move(h);
    }
    Handler* lookup(const std::string& svc, const std::string& name) {
        auto it = handlers.find(svc + "." + name);
        return it == handlers.end() ? nullptr : &it->second;
    }
private:
    std::unordered_map<std::string, Handler> handlers;
};

class RpcServer {
public:
    template <typename TArgs, typename TRet, typename Fn>
    void registerMethod(const std::string& svc, const std::string& name, Fn fn, std::shared_ptr<Codec> codec) {
        registry.registerMethod(svc, name, [fn, codec](const std::string& payload) {
            auto args = std::any_cast<TArgs>(codec->decode(payload, typeid(TArgs)));
            TRet out = fn(args);
            return codec->encode(out);
        });
    }
    RpcResponse handle(const RpcRequest& req) {
        RpcResponse res; res.id = req.id;
        auto* h = registry.lookup(req.serviceName, req.methodName);
        if (!h) { res.hasError = true; res.error = {"METHOD_NOT_FOUND", req.serviceName + "." + req.methodName}; return res; }
        try { res.result = (*h)(req.payload); }
        catch (const std::exception& e) { res.hasError = true; res.error = {"SERVER_ERROR", e.what()}; }
        return res;
    }
private:
    MethodRegistry registry;
};

TypeScript

type RpcError = { code: "METHOD_NOT_FOUND" | "BAD_ARGS" | "SERVER_ERROR" | "TIMEOUT"; message: string };
type RpcRequest = { id: string; serviceName: string; methodName: string; payload: unknown };
type RpcResponse = { id: string; result?: unknown; error?: RpcError };

interface Codec    { encode(o: unknown): unknown; decode<T>(b: unknown): T; }
interface Transport { send(req: RpcRequest): Promise<RpcResponse>; }

type AnyMethod = (...args: any[]) => any;

class MethodRegistry {
  private handlers = new Map<string, { instance: unknown; method: AnyMethod }>();

  register(serviceName: string, instance: Record<string, unknown>): void {
    const proto = Object.getPrototypeOf(instance);
    for (const name of Object.getOwnPropertyNames(proto)) {
      if (name === "constructor") continue;
      const fn = (instance as Record<string, unknown>)[name];
      if (typeof fn !== "function") continue;
      const key = `${serviceName}.${name}`;
      if (this.handlers.has(key)) throw new Error(`duplicate method: ${key}`);
      this.handlers.set(key, { instance, method: fn as AnyMethod });
    }
  }

  lookup(svc: string, name: string) { return this.handlers.get(`${svc}.${name}`); }
}

class RpcServer {
  private registry = new MethodRegistry();
  constructor(private codec: Codec) {}
  register(svc: string, inst: Record<string, unknown>) { this.registry.register(svc, inst); }

  async handle(req: RpcRequest): Promise<RpcResponse> {
    const h = this.registry.lookup(req.serviceName, req.methodName);
    if (!h) return { id: req.id, error: { code: "METHOD_NOT_FOUND", message: `${req.serviceName}.${req.methodName}` } };
    try {
      const args = this.codec.decode<unknown[]>(req.payload);
      const out = await h.method.apply(h.instance, args);
      return { id: req.id, result: this.codec.encode(out) };
    } catch (e: any) {
      return { id: req.id, error: { code: "SERVER_ERROR", message: String(e?.message ?? e) } };
    }
  }
}

class RpcClient {
  constructor(private transport: Transport, private codec: Codec, private defaultTimeoutMs = 1_000) {}

  async call<T>(svc: string, method: string, args: unknown[], timeoutMs?: number): Promise<T> {
    const req: RpcRequest = { id: crypto.randomUUID(), serviceName: svc, methodName: method, payload: this.codec.encode(args) };
    const sent = this.transport.send(req);
    const to = new Promise<RpcResponse>((_, rej) =>
      setTimeout(() => rej({ id: req.id, error: { code: "TIMEOUT", message: "timed out" } }), timeoutMs ?? this.defaultTimeoutMs)
    );
    const res = await Promise.race([sent, to]);
    if (res.error) throw res.error;
    return this.codec.decode<T>(res.result);
  }
}

// Loopback transport — same process, dispatches directly into the server.
class InMemoryTransport implements Transport {
  constructor(private server: RpcServer) {}
  send(req: RpcRequest): Promise<RpcResponse> { return this.server.handle(req); }
}

---

Extensibility

1. "Add a network transport (e.g., HTTP / gRPC)"

Implement Transport with a Netty/OkHttp/fetch client. Each send() POSTs an encoded RpcRequest to /rpc, multiplexed on a single connection with requestId correlation. Server side: a single endpoint that reads RpcRequest, calls server.handle, writes the response.

class HttpTransport implements Transport {
  constructor(private url: string) {}
  async send(req: RpcRequest): Promise<RpcResponse> {
    const r = await fetch(this.url, { method: "POST", body: JSON.stringify(req) });
    return await r.json();
  }
}

The cost: now you need framing, connection pooling, TLS, retry policy. None of that touches the registry, the server, or the codec — that is the value of the seam.

2. "Annotation-based service registration with method validation"

Add a @RpcMethod annotation. register() only picks up annotated methods, and validates that paramTypes and returnType are codec-serializable. Catches the "I forgot to mark this as RPC" bug at boot, not at first call.

@Retention(RetentionPolicy.RUNTIME) @Target(ElementType.METHOD) @interface RpcMethod {}

// in MethodRegistry.register:
for (Method m : ...) {
    if (!m.isAnnotationPresent(RpcMethod.class)) continue;
    // ... existing logic
}

3. "Streaming responses"

Today, Future<Response> resolves once. For server-streaming, change the client API to return a Flow<T> / AsyncIterable<T>. Server-side, the registered method returns an Iterable<T> or pushes via a callback. The transport must support multi-frame replies keyed on the same requestId.

The shape of the change: a new method RpcClient.stream(...), a new transport contract sendStream(req) -> Flow<Response>. The registry doesn't change.

4. "Interceptors / middleware (auth, tracing, retries)"

Wrap the RpcServer.handle and RpcClient.call in a chain of Interceptor objects — Decorator pattern. Each interceptor can short-circuit (auth fails) or annotate (tracing span). This is exactly how gRPC's ServerInterceptor / ClientInterceptor work.

---

What is Expected at Each Level

Level	Expectations
Mid (SMTS-track)	A working registry + call flow with reflection. Sync only is okay if explicit. Identifies Codec and Transport as separate concerns even if not fully implemented.
Senior / SMTS	Asynchronous CompletableFuture API, request-id multiplexing on a shared transport, error code mapping, timeouts. Codec/Transport pluggability behind interfaces. Discusses thread-safety of the registry and inflight map.
Staff	Backpressure on inflight map, cancellation propagation, interceptor chain (auth/trace/retry), service health and discovery seams, framing concerns when on a real wire, observability (per-method latency, error-rate, in-flight gauge). Knows when to switch from reflection to code-generated stubs.