Ein Port, drei Client-Crates — hexagonal in Rust

Notizen von der Werkbank. Ein Design-Durchgang, kein Tutorial: wie ein Rust-Service drei Blockchains über ein einziges Trait liest, warum dieses Trait genau diese Signatur hat und — der Teil, den die meisten Ports-and-Adapters-Texte auslassen — die drei Stellen, an denen wir bewusst aufgehört haben zu abstrahieren.

Das Schwierige an einer Datenplattform ist selten die Modellierung. Es sind die Grenzen: die Nahtstellen, an denen das SDK eines Anbieters auf Code trifft, den Sie pflegen müssen, und an denen dessen Vorstellung von einem „Block“ auf Ihre trifft. An diesen Nähten kommt Veränderung an, und sie an der falschen Stelle zu ziehen ist teuer — auf eine Weise, die bis zur zweiten Integration unsichtbar bleibt.

Dies ist ein durchgearbeitetes Beispiel für das Ziehen einer solchen Grenze — echtes Trait, echter Adapter-Code und die Überlegungen, die dazu geführt haben. Das System liest On-Chain-Daten von drei Chains über drei Client-Crates, geschrieben von Teams, die keinen Anlass haben, sich auf irgendetwas zu einigen. Wenn Sie Ingestion anderer Art betreiben — CDC-Konnektoren, Partner-APIs, IoT-Feeds, Marktdaten — sind die Chains nebensächlich und das Problem identisch: heterogene Quellen, eine Domäne, die nichts von ihnen wissen soll, und die Frage, wie weit man abstrahiert, bevor es mehr kostet als es einbringt.

Jedes Krypto-Observability-Projekt beginnt mit einer Chain. Jemand schreibt einen Ethereum-Indexer, der newHeads abonniert, Blöcke dekodiert und nach Kafka publiziert — und das funktioniert. Zwei Monate später kommt eine zweite Chain dazu, Solana, Polkadot oder ein L2, und die Druckfrage lautet, ob sie im selben Service lebt oder einen eigenen bekommt.

Derselbe Service ist meistens die ehrliche Antwort. Der Betriebsaufwand für zwei Ingester-Binaries — zwei Deployments, zwei Config-Oberflächen, zwei Dashboards, zwei On-Call-Runbooks — ist hoch, und nichts daran ist fundamental. Fundamental ist dagegen, dass jede Chain von einem anderen Rust-Crate mit völlig anderer Form bedient wird. Das Ethereum-Crate (alloy) reicht Ihnen ein WebSocket-Abonnement, das Header-Werte liefert. Das Solana-Crate (solana-client) zwingt Sie, zwei Clients nebeneinander zu betreiben — einen PubsubClient für Slot-Benachrichtigungen und einen RpcClient, um die Blöcke tatsächlich zu holen. Das Polkadot-Crate (subxt) liefert dekodierte Block<PolkadotConfig>-Werte, aber der Weg zu deren Extrinsics kostet zwei weitere Awaits.

Das sind drei wirklich unterschiedliche mentale Modelle, und die Reflexlösung — ein großes Chain-Enum, das überall im Ingester in chain-spezifische Codepfade verzweigt — bekommt mit jeder neuen Chain einen weiteren Arm und lässt schließlich die Idiome jedes Crates in jede darüberliegende Schicht sickern. Der bessere Zug ist ein Port: ein Rust-Trait, das alle Chain-Quellen implementieren, mit gerade genug Form, um die Schicht darüber zu komponieren, ohne zu wissen, welche Chain darunter liegt.

Was es tatsächlich hexagonal macht

Ports and Adapters wird gern als „benutze Interfaces an den Grenzen“ beschrieben, was zutrifft und nichts nützt — jede Codebasis hat Interfaces an Grenzen. Die eigentliche Regel ist enger, und sie betrifft die Richtung, in die die Abhängigkeiten zeigen.

Die Domäne definiert das Trait. Der Adapter implementiert es. Die Domäne importiert den Adapter nie.

Das kehrt die Richtung um, in die Abhängigkeiten von selbst fließen wollen. Ohne diese Umkehr entsteht Schichtung: die Anwendung importiert den Ethereum-Client, wickelt ihn in etwas ein, und die Form dieser Hülle wird davon bestimmt, was alloy zufällig nach außen gibt. Mit der Umkehr wird das Trait passend zum Use Case geschrieben, und jedes Anbieter-SDK muss sich verrenken, um hineinzupassen — was genau der Punkt ist, weil die Verrenkung dann auf eine Datei pro Anbieter beschränkt bleibt.

Das Diagramm in Worten: Die Chains treiben das System von oben, über Adapter, die je ein Anbieter-SDK umschließen und den ChainSource-Port implementieren. Der Kern reagiert und ruft dann über EventSink und MetricsSink nach unten zu Kafka und Prometheus. Die Pfeile zeigen Daten; die Abhängigkeiten laufen sämtlich in die Gegenrichtung, weil jeder dieser fünf Adapter gegen ein Trait geschrieben ist, das der Kern definiert — und der Kern nennt keinen von ihnen beim Namen. main.rs steht außerhalb des Bildes und verdrahtet die konkreten Adapter zur Konstruktionszeit.

In Rust wird das erzwungen, nicht bloß empfohlen. alloy taucht in genau einer Datei auf; würde das Anwendungsmodul es importieren, scheiterte der Build an einer nicht auflösbaren Abhängigkeit, weil der Crate-Graph es nicht zulässt. Das ist eine stärkere Garantie als eine Konvention, die im Review niemand prüft.

Warum ein treibender Port einen Channel zurückgibt

Hexagonal klassifiziert Ports danach, wer wen aufruft. Das ist der ganze Test, und er lohnt sich als Frage nach dem Akteur auf der anderen Seite — nicht nach der Richtung, in die die Daten zufällig fließen:

• Treibende (primäre) Ports werden von der Außenwelt aufgerufen. Dort draußen handelt etwas von sich aus — eine Anfrage trifft ein, ein Block wird produziert — und unsere Anwendung ist das, was in Bewegung gesetzt wird.
• Getriebene (sekundäre) Ports werden von der Anwendung aufgerufen. Der Akteur auf der anderen Seite tut nichts, bis unser Use Case nach ihm greift und ihn in Bewegung setzt. Eine Datenbank schreibt sich nicht selbst; Kafka entscheidet nicht, eine Nachricht anzunehmen.

Das Diagramm oben stapelt sie: treibend oben, getrieben unten. Cockburns ursprüngliches Hexagon und die meisten Darstellungen, denen Sie sonst begegnen, setzen treibend nach links und getrieben nach rechts — dieselbe Unterscheidung, um neunzig Grad gedreht.

Die Datenflussrichtung ist der falsche Test und liefert hier die falsche Antwort. EventSink bewegt Daten nach außen und ChainSource nach innen, aber das ist nicht, was sie trennt. Was sie trennt, ist wessen Aktivität überhaupt darüber entscheidet, dass Arbeit anfällt. Der Sink tut nichts, bis unser Use Case es ihm sagt. Die Chain wartet nicht auf uns — Blöcke entstehen, ob wir zuhören oder nicht, und unsere Aufgabe ist es, mit einem Akteur Schritt zu halten, den wir nicht kontrollieren.

Das ist bewusst keine Aussage darüber, wer auf Transportebene den Funktionsaufruf macht, denn die Antwort variiert dort und spielt keine Rolle. Unser Solana-Adapter pollt: er abonniert Slot-Benachrichtigungen und holt dann jeden Block per RPC, ein Aufruf pro Slot. Ein Backfill-Adapter würde einen ganzen Blockbereich pollen, ganz ohne Abonnement. Keiner von beiden ist ein getriebener Port.

Genau diesen Unterschied zu verbergen ist die Aufgabe des Adapters. Ob die Außenwelt zu uns hin drückt oder von uns gezogen werden muss, ist eine Eigenschaft des SDK eines Dritten. Ließe man das die Klassifikation des Ports ändern, verschöbe sich die Struktur der Anwendung jedes Mal, wenn ein Anbieter sein Zustellmodell ändert. Der Adapter pollt stellvertretend für einen Akteur, dessen Takt er nicht setzt — der Port bleibt so oder so treibend, und der Use Case darüber erfährt nie, welches von beidem es war.

ChainSource ist also ein treibender Port, und die Chain-Adapter sitzen auf der treibenden Seite, dort wo in einem anfragegetriebenen Service ein HTTP-Controller säße.

Interessant ist die Form, die er annimmt, denn der übliche treibende Adapter ruft in die Anwendung hinein. Ein HTTP-Handler nimmt eine Anfrage entgegen und ruft einen Use Case auf. Ein Message-Listener nimmt eine Nachricht entgegen und ruft einen Handler auf. In beiden Fällen Push, mit dem Adapter als Besitzer des Aufrufs.

Unserer kehrt das um: er gibt einen Receiver zurück und lässt die Anwendung daraus ziehen.

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

Diese Umkehr ist Absicht, und der Grund heißt Cancellation. Ein Push-basierter treibender Adapter besitzt die Kontrollschleife — er entscheidet, wann die Anwendung läuft. Dieser Service darf das nicht zulassen, weil er gleichzeitig auf eingehende Events und auf ein Shutdown-Signal warten muss — tokio::select!, das mehrere Futures gegeneinander laufen lässt und mit dem erstfertigen weitermacht. Ein Handler, der aus einem Adapter heraus aufgerufen wird, hat keinen Ort für diese Wahl. Der Besitz der Schleife muss bei der Anwendung bleiben.

Der Port behält also die Klassifikation (die Chain treibt) und kehrt die Zustellung um (die Anwendung zieht). Ausgeschrieben lautet der Tausch: Man gibt die Fähigkeit des Adapters auf, hineinzurufen, und gewinnt die Fähigkeit der Anwendung, Event-Verarbeitung mit Cancellation, Backpressure und allem anderen zu komponieren, worüber sie ein select! braucht.

Die Kosten landen beim Adapter, der nun selbst etwas laufen lassen muss, um diesen Channel zu füllen — einen Task, einen eigenen Client und eine Reconnect-Schleife. Das ist das Problem des nächsten Abschnitts, und dort hören die drei Crates auf, einander zu ähneln.

Welches Pattern welche Aufgabe übernimmt

Hier sind drei Patterns gestapelt, deren Namen oft genug austauschbar verwendet werden, dass es sich lohnt, sie festzunageln:

Hexagonal ist die Architektur; der GoF-Adapter ist eine verbreitete Art, dessen Adapter-Platz zu füllen. Adapter umschließt eine Anbieter-API, Bridge abstrahiert über eine Familie davon — und beide lassen sich kombinieren, denn ein Adapter an der Port-Grenze kann intern Bridge verwenden, was der Metrics-Sink genau tut.

Ein viertes Pattern taucht in der Schicht darüber auf: Fan-out (Task pro Quelle). Die Anwendung startet ein tokio::spawn pro Adapter und komponiert Cancellation über ein select!. Das ist nur ausdrückbar, weil jeder Adapter dasselbe Paar aus Channel und Handle zurückgibt — das Pattern über dem Port wird durch die am Port gewählte Form erst ermöglicht.

Der Port vollständig

#[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() ist ein Etikett, und is_healthy() bedient die Readiness-Probe. Alles bisher betraf subscribe(), und eine Alternative ist erwähnenswert, weil sie die Form ist, zu der die meisten Rust-Reviewer zuerst greifen.

Der Rückgabetyp des Ports — ein Channel und ein Cancel-Handle statt eines Streams — ist das, was drei wirklich verschiedene Client-Crates einstecken lässt, ohne dass die Anwendungsschicht weiß, welches sie gerade hat.

Warum nicht impl Stream? Es ist die idiomatische Rust-Antwort, es komponiert mit dem gesamten futures-Ökosystem — und es wäre hier falsch gewesen. Ein Stream trägt Werte; er trägt keine Möglichkeit, ihre Produktion zu beenden. Der Aufrufer müsste die Cancellation also selbst komponieren, und jeder Adapter bräuchte ein durch seine Interna gefädeltes Cancellation-Token, um einen Stream zu erzeugen, der sie respektiert. Diese Token-Verkabelung breitet sich durch die Trait-Hierarchie aus, jedes Crate implementiert Shutdown ein wenig anders, und die Einheitlichkeit, für die der Port existierte, ist dahin — drei Chains, die auf Typebene gleich aussehen und sich bei SIGTERM verschieden verhalten.

Indem er (mpsc::Receiver<_>, SourceHandle) zurückgibt, schiebt der Port das Cancellation-Problem in den Adapter hinein. Jeder Adapter startet seinen eigenen Task, besitzt seinen eigenen Client und übergibt dem Aufrufer zwei Dinge: einen Channel zum Lesen der Events und ein undurchsichtiges Handle, dessen Drop-Implementierung den Task beendet. Der Aufrufer fasst nie ein Cancellation-Token an; die Spawn- und Select-Interna des Adapters dringen nie nach oben.

Drei Adapter

Alle drei Adapter-Dateien beginnen gleich — Domain-Imports, Client-Crate-Imports, ein Struct und ein impl-Block — und laufen dann fast vollständig auseinander.

Jeder Ausschnitt beginnt mit den Anbieter-Imports der jeweiligen Datei, denn diese Zeilen sind die Grenze — jeder crate-spezifische Name im gesamten Service taucht in einem dieser drei Blöcke auf.

Ethereum (ethereum.rs) nutzt alloys WebSocket-Provider. Provider::subscribe_blocks() liefert ein Subscription<Header>; der Adapter leert diesen Stream, wandelt jeden Header in ein ChainEvent::BlockIngested und schiebt ihn in den 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) braucht zwei Clients nebeneinander. Es gibt kein kombiniertes Primitiv „abonniere ganze Blöcke“ — PubsubClient::slot_subscribe() liefert Slot-Benachrichtigungen, und für jeden Slot muss separat RpcClient::get_block_with_config() aufgerufen werden, um den eigentlichen Blockinhalt zu holen.

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) nutzt subxt 0.50, dessen Block-API sich im 0.50-Release in der Form geändert hat — client.stream_blocks() liefert standardmäßig finalisierte Blöcke, aber der Block-Header trägt den Zeitstempel nicht. Der muss aus dem Timestamp.Now-Storage geholt werden:

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
}

Drei Crates, drei Streaming-Modelle, drei Strategien zur Zeitstempel-Ermittlung. Was sie teilen, ist die Form an der Grenze: Jeder gibt aus subscribe() ein IngestorResult<(mpsc::Receiver<_>, SourceHandle)> zurück, und jeder startet seinen eigenen Task, der den Client besitzt und Events in den Channel schiebt.

Das Diagramm in Worten: Die Anwendungsschicht spricht mit ChainSource als Trait-Objekt; drei konkrete Adapter — jeder um ein anderes Client-Crate — implementieren es. Die Anwendung importiert nie ein chain-spezifisches Crate; die Typen jedes Crates bleiben in dessen eigener Adapter-Datei.

Der Abhängigkeitspfeil zeigt nur in eine Richtung. Das Anwendungsmodul importiert weder alloy noch solana-client noch subxt. Es könnte gar nicht kompilieren, wenn es das versuchte — diese Imports leben an genau einer Stelle pro Chain.

Die getriebene Seite, kurz

Bisher ging es durchweg um den treibenden Port, weil dort die drei Crates auseinandergehen. Die getriebenen Ports lohnen einen Blick zum Kontrast — sie sind anders geformt, und zwar aus Gründen, die aus der Klassifikation folgen und nicht aus Geschmack.

#[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 braucht weder Channel noch Cancel-Handle, weil er nach Request-Response funktioniert: die Anwendung ruft, der Adapter antwortet, der Aufruf endet. Das ist die gewöhnliche Form für einen getriebenen Port, und es ist die Form, die ChainSource nicht verwenden konnte — ein Abonnement hat keinen Punkt, an dem der Aufruf fertig ist.

MetricsSink ist noch einmal anders — synchron und ohne Rückgabewert. Der Port entfernt den Fehlerfall, statt eine Konvention darüber zu dokumentieren: Es gibt kein Result, das ein Aufrufer falsch behandeln könnte, und keinen Weg, auf dem ein Observability-Problem zu einem Ingestion-Problem wird.

Sein Cardinality-Contract ist den Platz wert, den er einnimmt. Prometheus speichert eine Zeitreihe pro eindeutiger Label-Kombination, sodass ein einziges unbeschränktes Label — eine Blocknummer, ein Transaktions-Hash — aus einem Zähler pro Block einen OOM-Kill auf dem Scrape-Target macht, Wochen später, ohne dass irgendein Test fehlschlägt. Das ist im Review unsichtbar, solange niemand aufgeschrieben hat, was „geschlossen“ bedeutet.

Ein Port ist ein guter Ort für eine Invariante, die jeden einzelnen Adapter überdauert. Das ist ein Nutzen der Grenze jenseits des Austauschens von Implementierungen — und derjenige, der am ehesten übersehen wird, wenn Ports and Adapters als „bloß Interfaces“ beschrieben werden.

Was der Port uns eingebracht hat

Sobald jeder Adapter einen Channel-Receiver zurückgibt, wird alles oberhalb des Ports einheitlich. Die Fan-out-Schicht, die alle drei Adapter nebenläufig betreibt, besteht aus einem tokio::spawn und einem select! pro Adapter, um Cancellation mit dem Empfang von Events zu komponieren — ganz ohne chain-spezifische Verzweigung.

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 und Backoff werden einheitlich, ohne geteilt zu sein. Weil jeder Adapter seine eigene Subscribe-Schleife besitzt, lebt die Reconnect-Logik — exponentielles Backoff, Maximalversuche, „sende SourceDisconnected auf den Channel, wenn die Versuche erschöpft sind“ — im Adapter und nicht in der Anwendung. Alle drei verwenden dieselbe Kadenz 2s → 4s → 8s → 64s, nicht weil der Port es erzwingt — es gibt keinen gemeinsamen Code zwischen den Subscribe-Schleifen — sondern weil die Festlegung, was „die Grenze“ bedeutet, das Muster trivial kopierbar gemacht hat.

Observability folgt derselben Form. Jeder Adapter ruft einen gemeinsamen Helfer observe_head_lag(...), der den Wert klemmt und dann MetricsSink::record_chain_head_lag_ms(chain, lag) aufruft — die Trait-Methode von oben. Eine Metrik, chain_head_lag_ms, unterschieden durch ihr chain-Label statt durch drei Metriknamen pro Chain: der Cardinality-Contract wird damit eingehalten und nicht bloß behauptet.

Was wir bewusst NICHT abstrahiert haben

Jede Abstraktion hat einen Fehlermodus, in dem sie nicht weiß, wann sie aufhören soll. Aus „Ports and Adapters“ wird schnell „ein gemeinsames Crate, das jeder Adapter importieren muss“, dann wird dieses Crate zu einem Ort, an dem man die Unterschiede der Client-Crates mit gemeinsamen Helfern übertüncht, und dann werden diese Helfer zum eigentlichen Interface — der Port ist Dekoration. Jeder der folgenden Punkte ist eine Stelle, an der wir aufgehört haben:

• Eine Adapter-Datei pro Chain, keine gemeinsamen Adapter-Helfer. ethereum.rs, solana.rs, polkadot.rs. Das ratenbegrenzte Warn-Atomic, das eine Uhrzeit-Drift-Warnung davon abhält, bei jedem Block zu feuern, existiert inzwischen in allen dreien — kopiert statt extrahiert. Das sind rund sieben duplizierte Zeilen pro Chain, und es erkauft die Gewissheit, dass eine Änderung an der Warn-Kadenz einer Chain nicht stillschweigend eine andere verändert. Bei drei Kopien ist das noch der richtige Tausch; es ist zugleich der Punkt, an dem uns eine vierte Chain zur erneuten Prüfung zwingen würde — was man besser ausspricht, als es später zu entdecken.
• Kein gemeinsames run_subscription_loop-Generic. Verlockend, denn die Form ist über alle drei hinweg nahezu identisch — aber „nahezu identisch“ verbirgt subxts Zwei-Await-Sprung für Extrinsics, Solanas separaten RpcClient::get_block_with_config-Aufruf pro Slot und Ethereums alloy-Eigenheit bei Subscription-IDs im Reconnect, um die herum die äußere Reconnect-Schleife eigens gebaut ist. Eine generische Schleife über impl Stream bräuchte entweder ein Config-Trait, das breiter ist als der Port selbst, oder sie schöbe diese Eigenheiten als Flags in den Adapter hinunter — schlechter als Duplikation.
• Die Domänenkonvertierung liegt pro Adapter. normalise_block_header, normalise_slot, normalise_block. Jede nimmt crate-spezifische Typen entgegen (alloys Header, Solanas UiConfirmedBlock, subxts Block<PolkadotConfig>) und erzeugt das domäneneigene ChainEvent. Diese Funktion ist die eigentliche Adapter-Grenze — die eine Stelle pro Chain, an der crate-spezifische Typen benannt werden. Sie generisch zu machen würde den ganzen Sinn des Patterns zunichtemachen.

Festzuhalten, was man nicht abstrahiert hat, ist das, was die Abstraktionsgrenze dort hält, wo man sie gezogen hat. Jeder dieser Punkte ist eine Stelle, an der weitere Extraktion für sich genommen vertretbar wäre, und jede hätte eine Schicht hinzugefügt, die durch Symmetrie statt durch eine Anforderung gerechtfertigt ist — und genau so hört ein Port auf, eine Grenze zu sein, und wird zu einer Abhängigkeit auf gemeinsamen Code mit einem Trait obendrauf.

Ehrliche Grenzen

Die Form „ein Spawn pro Chain“ trägt unterhalb von O(Dutzenden) Quellen. Jeder Adapter besitzt einen Tokio-Task, einen mpsc-Puffer und eine Verbindung, was bei vierzig günstig bleibt und bei vierhundert aufhört, günstig zu sein — etwa bei einem Adapter pro Rollup für jedes L2. Ab dort will man eine gemeinsame Event-Schleife mit Filtern pro Quelle, bezahlt mit gröberer Cancellation.

Der Port setzt außerdem voraus, dass jede Quelle ein Live-Abonnement ist. Historischer Backfill braucht einen zweiten Port auf derselben Seite — eine Bereichsquelle statt eines Streams — denn ChainSource auf beides zu dehnen hieße, ein Modus-Flag genau an die Stelle zu setzen, die das Design am härtesten sauber hält.

Zum Mitnehmen

Das Pattern unter alldem ist Ports and Adapters, eng angewandt: drei Traits im Besitz des Anwendungskerns, fünf Adapter, die sie implementieren, und ein Modul-Layout, das die Abhängigkeitsrichtung zu einem Compile-Fehler macht statt zu einer Konvention. Nichts Exotisches — der Wert kam daher, wo die Grenzen gezogen wurden und dass wir bewusst entschieden haben, wo sie aufhören.

Zwei Dinge, die sich unabhängig vom Stack mitnehmen lassen:

• Geben Sie einem treibenden Port über einer unbeschränkten Quelle einen Channel und ein Cancel-Handle, keinen Stream. Das ist die Form, mit der die Anwendung Cancellation und Backpressure komponieren kann, ohne ein Token durch jeden Adapter zu fädeln. Getriebene Ports brauchen das nicht — EventSink ist ein einfacher asynchroner Aufruf, weil ein Request endet.
• Halten Sie die crate-spezifischen Typen in einer Datei pro Quelle. Die Domänenkonvertierung ist die eigentliche Grenze: Alles darüber ist ein ChainEvent, alles darunter ist alloy::Header oder subxt::Block<PolkadotConfig> oder was das Client-Crate sonst herausgibt.

Sie zahlen ein Spawn und einen mpsc pro Quelle. Dafür ist die nächste Quelle eine neue Adapter-Datei, ein neuer Arm in einem Config-Enum und keine Änderung irgendwo oberhalb des Ports — ein Tausch, der sich bei jedem Ingester lohnt, dessen Quellen heterogen sind.

Wenn Sie diese Grenze anderswo gezogen haben — oder finden, dass ein Stream die richtige Wahl gewesen wäre, und begründen können, warum — würde ich gerne hören, wo sie gehalten oder nachgegeben hat. Die Tür steht offen unter cordata.tech/contact. Der Begleitbeitrag zur Observability-Seite desselben Service ist jetzt da: Die trace_id war der einfache Teil — was nötig war, damit eine Logzeile auf den richtigen Trace zeigt, und die vier Bedingungen, die dafür erfüllt sein mussten.