One port, three client crates — a hexagonal Rust ingester

Notes from the workbench. A design walkthrough rather than a tutorial: how one Rust service reads three blockchains through a single trait, why that trait has the signature it does, and — the part most ports-and-adapters write-ups skip — the three places we deliberately stopped abstracting.

Most of what makes a data platform hard is not the modelling. It is the boundaries: the seams where someone else's SDK meets code you have to maintain, and where a vendor's idea of a "block" meets yours. Those seams are where change arrives, and drawing them in the wrong place is expensive in a way that stays invisible until the second integration.

This is a worked example of drawing one — real trait, real adapter code, and the reasoning that produced them. The system ingests on-chain data from three chains through three client crates written by teams with no reason to agree on anything. If you work on ingestion of any other kind — CDC connectors, partner APIs, IoT feeds, market data — the chains are incidental and the problem is identical: heterogeneous sources, a domain that should not know about them, and the question of how much to abstract before it costs more than it saves.

Every crypto observability project starts with one chain. Someone writes an Ethereum indexer that subscribes to newHeads, decodes blocks and publishes to Kafka, and it does the job. Two months later a second chain arrives — Solana, Polkadot, an L2 — and the pressure question is whether it lives in the same service or gets its own.

The same service is usually the honest answer. The operational cost of running two ingester binaries — two deployments, two config surfaces, two dashboards, two on-call runbooks — is high, and none of it is fundamental. What is fundamental is that each chain is served by a different Rust crate with a completely different shape. The Ethereum crate (alloy) hands you a WebSocket subscription that yields Header values. The Solana crate (solana-client) makes you run two clients side by side — a PubsubClient for slot notifications and an RpcClient for actually fetching blocks. The Polkadot crate (subxt) yields decoded Block<PolkadotConfig> values, but reaching their extrinsics costs two further awaits.

Those are three genuinely different mental models, and the reflex fix — a big Chain enum branching to per-chain code paths everywhere the ingester runs — grows a new arm each time a chain is added, and eventually leaks each crate's idioms into every layer above it. The better move is a port: one Rust trait that all chain sources implement, with just enough shape to compose the layer above without knowing which chain is downstream.

What actually makes it hexagonal

Ports and adapters gets described as "use interfaces at the boundaries", which is true and useless — every codebase has interfaces at boundaries. The actual rule is narrower, and it is about which direction the dependencies point.

The domain defines the trait. The adapter implements it. The domain never imports the adapter.

That inverts the direction dependencies naturally want to flow. Without inversion you get layering: the application imports the Ethereum client, wraps it in something, and the wrapper's shape is decided by whatever alloy happens to expose. With inversion the trait is written to suit the use case, and each vendor SDK has to contort itself to fit — which is the point, because the contortion is then confined to one file per vendor.

The diagram, in words: chains drive the system from the top, through adapters that wrap each vendor SDK and implement the ChainSource port. The core reacts, then calls outward through EventSink and MetricsSink to Kafka and Prometheus below. The arrows show data; the dependencies all run the other way, because every one of those five adapters is written against a trait the core defines, and the core names none of them. main.rs sits outside the picture and wires the concrete adapters in at construction time.

In Rust this is enforced rather than encouraged. alloy appears in exactly one file; if the application module tried to import it, the build would fail on an unresolved dependency, because the crate graph does not permit it. That is a stronger guarantee than a convention nobody checks at review time.

Why a driving port hands back a channel

Hexagonal classifies ports by who invokes whom. That is the whole test, and it is worth stating as a question about the actor on the other side rather than about the direction data happens to move:

• Driving (primary) ports are invoked by the outside world. Something out there acts on its own — a request arrives, a block is produced — and our application is what gets kicked into behaviour.
• Driven (secondary) ports are invoked by the application. The actor on the other side does nothing until our use case reaches out and kicks it into behaviour. A database does not write itself; Kafka does not decide to accept a message.

The diagram above stacks them: driving at the top, driven at the bottom. Cockburn's original hexagon and most renderings you will meet elsewhere put driving on the left and driven on the right — the same distinction, turned ninety degrees.

Data-flow direction is the wrong test and gives the wrong answer here. EventSink moves data outward and ChainSource moves it inward, but that is not what separates them. What separates them is whose activity decides that work happens at all. The sink does nothing until our use case tells it to. The chain does not wait for us — blocks are produced whether we are listening or not, and our job is to keep up with an actor we do not control.

That is deliberately not a claim about who makes the function call at the transport layer, because the answer there varies and does not matter. Our Solana adapter polls: it subscribes to slot notifications and then fetches each block over RPC, one call per slot. A backfill adapter would poll a whole block range with no subscription at all. Neither is a driven port.

Hiding that difference is exactly what the adapter is for. Whether the external world pushes at us or has to be pulled from is a property of someone else's SDK, and letting it change the port's classification would mean the application's structure shifts every time a vendor changes their delivery model. The adapter polls on behalf of an actor whose schedule it does not set — the port stays driving either way, and the use case above it never learns which it was.

So ChainSource is a driving port, and the chain adapters sit on the driving side where an HTTP controller would in a request-driven service.

The interesting part is the shape it takes, because the usual driving adapter calls into the application. An HTTP handler receives a request and invokes a use case. A message listener receives a message and invokes a handler. Push, in both cases, with the adapter owning the call.

Ours inverts that, handing back a receiver and letting the application pull from it:

async fn subscribe(&self) -> IngestorResult<(
    mpsc::Receiver<Result<ChainEvent, IngestorError>>,
    SourceHandle,
)>;

That inversion is deliberate, and cancellation is the reason. A push-based driving adapter owns the control loop — it decides when the application runs. This service cannot allow that, because it has to wait on incoming events and a shutdown signal at the same time — tokio::select!, which races several futures and proceeds with whichever finishes first. A handler being called from inside an adapter has nowhere to put that choice. Ownership of the loop has to stay with the application.

So the port keeps the classification (the chain drives) and inverts the delivery (the application pulls). Written out, the trade is: give up the adapter's ability to call in, gain the application's ability to compose event handling with cancellation, backpressure and anything else it needs to select! over.

The cost lands on the adapter, which now has to run something of its own to fill that channel — a task, a client it owns, and a reconnect loop. Which is the next section's problem, and where the three crates stop resembling each other.

Which pattern is doing which job

Three patterns are stacked here, and their names get used interchangeably often enough to be worth pinning down:

Hexagonal is the architecture; the GoF Adapter is one common way to fill its adapter slot. Adapter wraps one vendor API, Bridge abstracts over a family of them — and the two compose, because an adapter at the port boundary can use Bridge internally, which is exactly what the metrics sink does.

A fourth pattern shows up at the layer above: fan-out (task-per-source). The application runs one tokio::spawn per adapter and composes cancellation with a select!. That is only expressible because every adapter returns the same channel-and-handle pair — the pattern above the port is enabled by the shape chosen at it.

The port in full

#[async_trait]
pub trait ChainSource: Send + Sync + 'static {
    fn chain(&self) -> Chain;
 
    async fn subscribe(
        &self,
    ) -> IngestorResult<(
        mpsc::Receiver<Result<ChainEvent, IngestorError>>,
        SourceHandle,
    )>;
 
    async fn is_healthy(&self) -> bool;
}

chain() is a tag and is_healthy() backs the readiness probe. Everything above concerned subscribe(), and there is one alternative worth addressing, because it is the shape most Rust reviewers reach for first.

The port's return type — a channel and a cancel handle rather than a stream — is what lets three genuinely different client crates plug in without the application layer knowing which one it has.

Why not impl Stream? It is the idiomatic Rust answer, it composes with the whole futures ecosystem, and it would have been wrong here. A stream carries values; it carries no way to stop producing them. So the caller would have to compose cancellation itself and every adapter would need a cancellation token threaded through its internals to produce a stream that honours it. That token-plumbing spreads through the trait hierarchy, each crate ends up implementing shutdown slightly differently, and the uniformity the port existed to provide is gone — three chains that look alike at the type level and behave differently on SIGTERM.

By returning (mpsc::Receiver<_>, SourceHandle), the port pushes the cancellation problem inside the adapter. Each adapter spawns its own task, owns its own client, and hands the caller two things: a channel to read events from, and an opaque handle whose Drop implementation shuts the task down. The caller never touches a cancellation token; the adapter's spawn/select internals never leak upward.

Three adapters

All three adapter files start the same way — domain imports, client-crate imports, a struct and an impl block — and then diverge almost completely.

Each snippet opens with that file's vendor imports, because those lines are the boundary — every crate-specific name in the whole service appears in one of these three blocks.

Ethereum (ethereum.rs) uses alloy's WebSocket provider. Provider::subscribe_blocks() returns a Subscription<Header>; the adapter drains that stream, converts each header into a ChainEvent::BlockIngested, and pushes it to the mpsc.

use alloy::{
    providers::{Provider, ProviderBuilder},
    rpc::types::Header,
    transports::{ws::WsConnect, TransportError},
};
 
let ws = WsConnect::new(&ws_url);
let provider = ProviderBuilder::new().connect_ws(ws).await?;
let sub = provider.subscribe_blocks().await?;
let mut stream = sub.into_stream();
 
while let Some(header) = stream.next().await {
    let event = normalise_block_header(header);
    event_tx.send(Ok(event)).await?;
}

Solana (solana.rs) needs two clients running side by side. There is no combined "subscribe to full blocks" primitive — PubsubClient::slot_subscribe() gives you slot notifications, and for each slot you have to call RpcClient::get_block_with_config() separately to fetch the actual block body.

use solana_client::{
    nonblocking::{pubsub_client::PubsubClient, rpc_client::RpcClient},
    rpc_config::{CommitmentConfig, RpcBlockConfig},
    rpc_response::SlotInfo,
};
use solana_transaction_status_client_types::{
    TransactionDetails, UiConfirmedBlock, UiTransactionEncoding,
};
 
// slot_subscribe hands back two things: the stream and an
// unsubscribe callback. The callback is dropped — a subscription here
// ends by dropping the client along with it.
let (mut slot_notifications, _unsubscribe) = pubsub_client.slot_subscribe().await?;
 
while let Some(slot_info) = slot_notifications.next().await {
    // second client, second round trip — the slot notification carries
    // no block body
    let block = rpc_client
        .get_block_with_config(slot_info.slot, rpc_block_config)
        .await?;
    let event = normalise_slot(slot_info.slot, block, Utc::now());
    event_tx.send(Ok(event)).await?;
}

Polkadot (polkadot.rs) uses subxt 0.50, whose block API changed shape in the 0.50 release — client.stream_blocks() yields finalized blocks by default, but the block header doesn't carry the timestamp. That has to be pulled from Timestamp.Now storage:

use subxt::{client::Block, OnlineClient, PolkadotConfig};
use subxt::rpcs::{client::RpcClient, methods::LegacyRpcMethods};
 
let client = OnlineClient::<PolkadotConfig>::from_url(&ws_url).await?;
let mut block_stream = client.stream_blocks().await?;
 
while let Some(block) = block_stream.next().await {
    let block = block?;
    let at_block = block.at().await?;
    let timestamp_addr = subxt::dynamic::storage::<(), u64>("Timestamp", "Now");
    let ms: u64 = at_block.storage().fetch(timestamp_addr, ()).await?.decode()?;
    // …build ChainEvent::BlockIngested with a real block_time
}

Three crates, three streaming models, three timestamp-extraction strategies. What they share is the shape at the boundary: each returns an IngestorResult<(mpsc::Receiver<_>, SourceHandle)> from subscribe(), and each spawns its own task that owns the client and pushes events into the channel.

The diagram, in words: the application layer talks to ChainSource as a trait object; three concrete adapters — each wrapping a different client crate — implement it. The application never imports a chain-specific crate; every crate's types stay inside its own adapter file.

The dependency arrow only points one way. The application module doesn't import alloy, solana-client, or subxt. It couldn't compile if it tried — those imports live in exactly one place per chain.

The driven side, briefly

Everything so far has been the driving port, because that is where the three crates disagree. The driven ports are worth a look for contrast — they are shaped differently, and for reasons that follow from the classification rather than from taste.

#[async_trait]
pub trait EventSink: Send + Sync + 'static {
    async fn publish_one(&self, event: ChainEvent) -> IngestorResult<()>;
    async fn flush(&self) -> IngestorResult<()>;
}
 
/// Outbound port: observability counters.
///
/// Infallible by design — recording a metric must never fail an ingestion.
///
/// # Cardinality contract
///
/// The label set used by implementations of this trait is closed: the only
/// permitted label dimension is `chain`. Implementations must not introduce
/// additional dimensions, and new methods must not take parameters carrying
/// unbounded values — block numbers, transaction hashes, wallet addresses,
/// log indices.
///
/// The rule lives at the port, not at the adapter, because future backends
/// (OTLP-metrics, vendor SaaS) inherit it via this trait. Enforcement is
/// code review at the call site: there is no compile-time guard, and a
/// panicking sink would turn a slow cardinality bug into a fast process kill.
pub trait MetricsSink: Send + Sync + 'static {
    fn record_block_ingested(&self, chain: Chain);
    fn record_event_published(&self, chain: Chain);
    fn record_chain_head_lag_ms(&self, chain: Chain, lag_ms: i64);
    // ...
}

EventSink needs neither a channel nor a cancel handle, because it is request-and-response: the application calls, the adapter answers, and the call ends. That is the ordinary shape for a driven port, and it is the shape ChainSource could not use — a subscription has no point at which the call is finished.

MetricsSink is different again — synchronous, returning nothing at all. The port removes the failure case rather than documenting a convention about it: there is no Result for a caller to mishandle, and no way for an observability problem to become an ingestion problem.

Its cardinality contract is worth the space it takes. Prometheus stores one time series per unique label combination, so a single unbounded label — a block number, a transaction hash — turns a per-block counter into an OOM kill on the scrape target, weeks later, with nothing failing in a test. That is invisible at review time unless someone has written down what "closed" means.

A port is a good place to put an invariant that outlives any particular adapter. That is a use for the boundary beyond swapping implementations, and the one most likely to be missed when ports and adapters get described as "just interfaces".

What the port bought us

Once every adapter returns a channel receiver, everything above the port becomes uniform. The fan-out layer that runs all three adapters concurrently is one tokio::spawn and one select! per adapter to compose cancellation with event reception, with no chain-specific branching anywhere in it.

for source in chain_sources {
    let (mut events, _handle) = source.subscribe().await?;
    let chain = source.chain();
 
    tokio::spawn(async move {
        loop {
            tokio::select! {
                biased;
                _ = &mut cancel_rx => break,
                event = events.recv() => match event {
                    Some(Ok(e))  => publish_to_kafka(chain, e).await,
                    Some(Err(e)) => log_and_reconnect(chain, e).await,
                    None         => break,
                }
            }
        }
    });
}

Reconnect and backoff end up uniform without being shared. Because each adapter owns its own subscribe loop, the reconnect logic — exponential backoff, max-attempts, "send SourceDisconnected on the channel when retries exhaust" — lives inside the adapter, not in the application. All three use the same 2s → 4s → 8s → 64s cadence, not because the port enforces it — there is no shared code between the subscribe loops — but because fixing what "the boundary" meant made the pattern trivial to copy.

Observability follows the same shape. Each adapter calls a shared observe_head_lag(...) helper, which clamps the value and then calls MetricsSink::record_chain_head_lag_ms(chain, lag) — the trait method above. One metric, chain_head_lag_ms, distinguished by its chain label rather than by three per-chain metric names, which is the cardinality contract being honoured rather than merely stated.

What we deliberately did NOT abstract

Every abstraction has a failure mode where it doesn't know when to stop. "Ports and adapters" easily becomes "shared crate that every adapter must import", and then the shared crate becomes a place where you paper over the client crates' differences with common helpers, and then the helpers become the actual interface — the port is decorative. Each of the following is a place we stopped:

• One adapter file per chain, no shared adapter helpers. ethereum.rs, solana.rs, polkadot.rs. The rate-limited-warn atomic that keeps a clock-skew warning from firing every block now exists in all three, copied rather than extracted. That is roughly seven duplicated lines per chain, and it buys the certainty that changing one chain's warn cadence cannot silently change another's. At three copies this is still the right trade; it is also the point at which a fourth chain would make us re-examine it, which is worth saying out loud rather than discovering later.
• No shared run_subscription_loop generic. Tempting — the shape is nearly identical across the three — but "nearly identical" hides subxt's two-await-hop for extrinsics, Solana's separate RpcClient::get_block_with_config call per slot, and Ethereum's alloy subscription-id reconnect quirk that the outer reconnect loop is written specifically to route around. A generic loop over impl Stream would either need a config trait wider than the port itself, or it would push those idiosyncrasies down into the adapter as flags — worse than duplication.
• The domain conversion is per-adapter. normalise_block_header, normalise_slot, normalise_block. Each takes crate-specific types (alloy's Header, solana's UiConfirmedBlock, subxt's Block<PolkadotConfig>) and produces the domain-owned ChainEvent. This function is the actual adapter boundary — the one place per chain where crate-specific types are named. Making it generic would defeat the entire point of the pattern.

Recording what you did not abstract is what keeps the abstraction boundary where you put it. Each of those is a point where further extraction is defensible in isolation, and each would have added a layer justified by symmetry rather than by a requirement — which is how a port stops being a boundary and becomes a shared-code dependency with a trait on top.

Honest limits

The one-spawn-per-chain shape holds below O(dozens) of sources. Each adapter owns a Tokio task, an mpsc buffer and a connection, which stays cheap at forty and stops being cheap at four hundred — a per-rollup adapter for every L2, say. At that point you want a shared event loop with per-source filters, paid for in coarser cancellation.

The port also assumes every source is a live subscription. Historical backfill needs a second port on the same side — a range source rather than a stream — because stretching ChainSource to cover both would put a mode flag in the one place the design works hardest to keep clean.

The take-home

The pattern underneath all of this is ports and adapters, applied narrowly: three traits owned by the application core, five adapters implementing them, and a module layout that makes the dependency direction a compile error rather than a convention. Nothing exotic — the value came from where the boundaries were drawn and from being deliberate about where they stopped.

Two things worth stealing, independent of the stack:

• Give a driving port over an unbounded source a channel and a cancel handle, not a stream. That is the shape that lets the application compose cancellation and backpressure without a token threaded through every adapter. Driven ports do not need it — EventSink is a plain async call, because a request ends.
• Keep the crate-specific types inside one file per source. The domain conversion function is the real boundary: everything above it is a ChainEvent, everything below it is alloy::Header or subxt::Block<PolkadotConfig> or whatever the client crate hands you.

You pay one spawn and one mpsc per source. In exchange, adding the next one is a new adapter file, a new arm in a config enum, and no change anywhere above the port — a trade worth taking on any ingester whose sources are heterogeneous.

If you have drawn this boundary somewhere different — or think returning a stream would have been the right call and can say why — I would genuinely like to hear where it held or bent. The door is open at cordata.tech/contact. The companion post on the observability side of this same service is now up: The trace_id was the easy part — what it took to make a log line point at the right trace, and the four things that had to be true before it meant anything.