Wire Proxy — Fault Injection & Wire-Level Monitoring

Morphium's test sources ship a small, reusable TCP proxy for the MongoDB wire protocol: WireProxy. It sits between any wire-protocol client (Morphium, the official drivers, mongosh) and any wire-protocol backend (MongoDB or PoppyDB) and gives you three things that are otherwise hard to get in a test:

1. Fault injection — freeze, reset, or cleanly close connections at runtime, without touching the server process. This is how DriverFailoverProxyTest reproduces failovers (clean stepdown, hard kill, frozen socket) in the normal CI matrix, with no kill -9 and no hand-built infrastructure.
2. Wire-level monitoring — observe every server response frame as a parsed WireProtocolMessage, e.g. to log exactly what a server sends during a test, or to assert on protocol-level behavior your API-level test can't see.
3. Response manipulation — rewrite server replies before the client sees them: change topology information (that is how the failover suite works), mutate documents, or inject deliberately malformed replies to test client robustness.

It lives in morphium-core's test sources — package de.caluga.test.morphium.testutil.proxy — so it is available to every test in this repository. It is not (yet) published as a standalone artifact; if you want to use it outside this repo, copy the package (it has no dependencies beyond WireProtocolMessage) or open an issue.

Quick start

// Proxy in front of any wire-protocol server (MongoDB or PoppyDB)
WireProxy proxy = new WireProxy("localhost", 27017);
proxy.addObserver(new Slf4jFrameObserver());   // log every server response frame
proxy.start();

// Point the client at the proxy, not the server
MorphiumConfig cfg = new MorphiumConfig();
cfg.clusterSettings().setHostSeed("localhost:" + proxy.getListenPort());

// ... run the test ...

proxy.stop();   // severs all connections, joins every pump thread before returning

WireProxy implements AutoCloseable, so try-with-resources works too. stop() guarantees that every internal pump thread has exited before it returns — no thread leakage across tests.

Full example: 3-node replica set, everything logged

The fragments above show single pieces; this is the whole thing end to end — three proxies in front of a three-node replica set, address rewriting so the driver never escapes the proxies, an observer logging every server reply, one write and one read flowing through, and a clean teardown. Runs as-is from morphium-core's test scope (that's where WireProxy and UncachedObject live):

import java.util.*;

import de.caluga.morphium.Morphium;
import de.caluga.morphium.MorphiumConfig;
import de.caluga.morphium.driver.wireprotocol.OpMsg;
import de.caluga.morphium.driver.wireprotocol.WireProtocolMessage;
import de.caluga.test.mongo.suite.data.UncachedObject;
import de.caluga.test.morphium.testutil.proxy.AddressRewriter;
import de.caluga.test.morphium.testutil.proxy.WireProxy;

public class WireProxyDemo {

    public static void main(String[] args) throws Exception {
        // The RS members EXACTLY as the servers report them in hello ("Map-key invariant":
        // if the server says "mongo1:27017", the key must be "mongo1:27017", not an IP).
        List<String> members = List.of("mongo1:27017", "mongo2:27017", "mongo3:27017");

        // 1) One proxy per member, each on a random local port.
        List<WireProxy> proxies = new ArrayList<>();
        Map<String, String> backendToProxy = new LinkedHashMap<>();
        for (String member : members) {
            String host = member.substring(0, member.indexOf(':'));
            int port = Integer.parseInt(member.substring(member.indexOf(':') + 1));
            WireProxy proxy = new WireProxy(host, port);
            proxies.add(proxy);
            backendToProxy.put(member, "localhost:" + proxy.getListenPort());
        }

        // 2) One shared rewriter (so every hello reply, from every node, maps the full
        //    topology to proxy addresses) + a logging observer on every proxy.
        AddressRewriter rewriter = new AddressRewriter(backendToProxy);
        for (WireProxy proxy : proxies) {
            proxy.setRewriter(rewriter);
            proxy.addObserver((dir, msg, ctx) ->
                System.out.printf("[proxy:%d] %s %s%n", ctx.listenPort(), dir, summarize(msg)));
            proxy.start();
        }
        System.out.println("topology mapping: " + backendToProxy);

        // 3) Morphium gets ONLY the proxy addresses as its seed. SSL and wire compression
        //    must be OFF - the proxy cannot frame-parse either (deliberate non-goal).
        MorphiumConfig cfg = new MorphiumConfig();
        cfg.connectionSettings().setDatabase("wireproxy_demo");
        cfg.clusterSettings().getHostSeed().clear();
        backendToProxy.values().forEach(cfg.clusterSettings()::addHostToSeed);
        cfg.driverSettings().setDriverName("PooledDriver");
        cfg.clusterSettings().setHeartbeatFrequency(1000);
        cfg.driverSettings().setServerSelectionTimeout(5000);
        cfg.connectionSettings().setUseSSL(false);
        cfg.driverSettings().setCompressionType(MorphiumConfig.CompressionType.NONE);

        // 4) Everything from here on - discovery hellos, heartbeats, the write, the read -
        //    shows up line by line in the observer output.
        try (Morphium morphium = new Morphium(cfg)) {
            morphium.store(new UncachedObject("hello through the proxy", 42));
            long count = morphium.createQueryFor(UncachedObject.class).countAll();
            System.out.println("read back through the proxies: " + count + " document(s)");

            // Optional: watch the driver cope with a frozen node. Freeze the first proxy -
            // its connections go silent (no error, no close), exactly like a paused VM.
            // proxies.get(0).setFaultMode(FaultMode.freeze);
        } finally {
            // Severs every connection (hard RST) and joins all pump threads before returning.
            for (WireProxy proxy : proxies) {
                proxy.stop();
            }
        }
    }

    /** Compact one-liner per frame: hello replies show the (rewritten!) topology,
     *  everything else just its top-level keys. */
    private static String summarize(WireProtocolMessage msg) {
        if (msg instanceof OpMsg op && op.getFirstDoc() != null) {
            Map<String, Object> doc = op.getFirstDoc();
            if (doc.containsKey("hosts")) {
                return "hello(primary=" + doc.get("primary") + ", hosts=" + doc.get("hosts") + ")";
            }
            return "OpMsg" + doc.keySet();
        }
        return msg.getClass().getSimpleName();
    }
}

Typical output — note that the hello lines already show proxy addresses, which is the address rewriting doing its job; if a real backend address ever shows up here, your backendToProxy keys don't match what the server reports:

topology mapping: {mongo1:27017=localhost:52114, mongo2:27017=localhost:52115, mongo3:27017=localhost:52116}
[proxy:52114] BACKEND_TO_CLIENT hello(primary=localhost:52114, hosts=[localhost:52114, localhost:52115, localhost:52116])
[proxy:52115] BACKEND_TO_CLIENT hello(primary=localhost:52114, hosts=[localhost:52114, localhost:52115, localhost:52116])
[proxy:52114] BACKEND_TO_CLIENT OpMsg[n, electionId, opTime, ok, ...]
[proxy:52114] BACKEND_TO_CLIENT OpMsg[cursor, ok, ...]
read back through the proxies: 1 document(s)

Fault injection

Faults are switched at runtime via proxy.setFaultMode(...):

FaultMode	Existing connections	New connection attempts	Simulates
passthrough	forwarded normally	accepted	healthy network (default)
freeze	left open, never answered, never closed	accepted, then silence	frozen process (kill -STOP), network partition, paused VM — the failure a client cannot distinguish from a slow server
reset	severed with a hard RST	refused	dead process (kill -9), closed port
close	severed with a clean FIN	refused	this route to the node is gone (the node itself may live on — e.g. after a clean stepdown)

Semantics worth knowing before you build a test on them:

• freeze is one-way per connection. A connection accepted (or already open) during a freeze stays parked until stop() — switching back to passthrough only affects connections opened after the switch. That mirrors reality: a socket to a frozen process does not spring back to life; the client has to time out and reconnect.
• close and reset both refuse new connections — they differ only in how existing ones are severed (FIN vs. RST). "Nothing reachable behind this proxy right now" is the contract.
• stop() always severs with RST, regardless of the configured fault mode, so a client blocked in a read sees a definite error rather than a clean EOF that would look like an orderly shutdown.

Replica sets: the address-rewriting trick

A proxy per node is not enough for a replica set: drivers do server discovery via hello, and the server answers with the real addresses (hosts, primary, me). After the first hello, a driver would connect straight past your proxies.

AddressRewriter fixes that. It is a ResponseRewriter that detects hello/isMaster-shaped replies structurally (setName + hosts present) and maps every real address to its proxy address:

// one proxy per RS member
Map<String, String> backendToProxy = Map.of(
    "mongo1:27017", "localhost:" + p1.getListenPort(),
    "mongo2:27017", "localhost:" + p2.getListenPort(),
    "mongo3:27017", "localhost:" + p3.getListenPort());

AddressRewriter rewriter = new AddressRewriter(backendToProxy);
p1.setRewriter(rewriter);
p2.setRewriter(rewriter);
p3.setRewriter(rewriter);

The driver now lives in a consistent alternate topology consisting entirely of proxies — every connection it ever opens, including discovery-triggered ones, flows through a fault gate. The map keys must be the exact host:port strings the server reports (watch out for hostname vs. IP mismatches). DriverFailoverProxyTest.assertOnlyConnectedThroughProxies shows how to verify no traffic leaks around the proxies.

To drive the real replica set while the driver only sees proxies (e.g. trigger a genuine replSetStepDown, poll replSetGetStatus), use ControlChannel — an auth-aware direct connection to the real nodes, deliberately separate from the proxied data path.

Monitoring: FrameObserver

proxy.addObserver((dir, msg, ctx) ->
    log.info("[{}] {} -> {}", ctx.listenPort(), dir, msg.getClass().getSimpleName()));

Observers are read-only by contract — they must not mutate the frame (rewriting is ResponseRewriter's job; the interfaces are deliberately separate: fault = state, rewrite = strategy, observe = listener). Slf4jFrameObserver is a ready-made logging implementation. Observer exceptions never kill a proxy thread.

Today only BACKEND_TO_CLIENT frames fire: client→backend traffic is forwarded as raw, length-prefixed bytes without parsing — deliberate pass-through fidelity, the proxy cannot distort what it does not interpret. The CLIENT_TO_BACKEND direction exists in the enum and is reserved for a consumer that actually needs it.

Injecting invalid or manipulated replies

ResponseRewriter receives every parsed server reply and returns what the client should see — including something intentionally broken:

proxy.setRewriter(reply -> {
    if (reply instanceof OpMsg msg && msg.getFirstDoc() != null
            && msg.getFirstDoc().containsKey("cursor")) {
        msg.getFirstDoc().put("ok", 0.0);            // flip a find reply into an error
        msg.getFirstDoc().put("errmsg", "injected"); // ... or corrupt it any way you like
    }
    return reply;
});

This is the hook for robustness tests: truncated cursors, unexpected error codes, protocol violations, replies claiming a different topology than reality. Only the backend→client direction can be rewritten — requests pass through untouched by design.

Limitations (honest list)

• No latency injection — a frame is forwarded immediately or not at all. If you need slow-link simulation, that would be a new FaultMode.
• No request (client→backend) rewriting or observation — see above, deliberate.
• freeze is not reversible per connection (matches reality, but don't expect a parked connection to resume).
• It is test infrastructure: no TLS termination, no config file, one backend per proxy instance.

Reference consumer

DriverFailoverProxyTest (tag wire-failover) is the full-scale example: three proxies in front of a real replica set, address rewriting, freeze/reset/close scenarios, stepdown via ControlChannel, and read/write/messaging recovery assertions. It runs in the normal test matrix against both MongoDB and PoppyDB — see the Developer Testing Guide for how the tags fit together.