feat(solana-indexer): PR 4 — Component structs (skeleton declarations)

Cargo.toml

@@ -149,6 +149,7 @@ tracing-subscriber = { version = "0.3.22", features = ["json"] }
 url = "2.5.0"
 vergen = "8"
 winner-selection = { path = "crates/winner-selection" }
+yellowstone-grpc-client = "13.1.0"
 yellowstone-grpc-proto = { version = "12.4.0", default-features = false }
 
 [workspace.lints]

crates/solana-indexer/Cargo.toml

@@ -18,11 +18,14 @@ path = "src/main.rs"
 [dependencies]
 async-trait = { workspace = true }
 bytes = { workspace = true }
+dashmap = { workspace = true }
 derive_more = { workspace = true }
 solana-client = { workspace = true }
 solana-sdk = { workspace = true }
 thiserror = { workspace = true }
+tokio = { workspace = true, features = ["macros", "rt-multi-thread", "sync"] }
 tracing = { workspace = true }
+yellowstone-grpc-client = { workspace = true }
 yellowstone-grpc-proto = { workspace = true }
 
 [lints]

crates/solana-indexer/src/indexer/decoder.rs

@@ -0,0 +1,71 @@
+#![expect(dead_code)]
+//! The decoder pulls `StreamUpdate`s from the ingester, decodes
+//! settlement-program and SolFlow transactions, joins account-update snapshots,
+//! and persists typed events.
+
+// TODO: This file only declares the component skeleton. The `run` body is
+// `unimplemented!`; the dispatch logic and persist path arrive in a later
+// change.
+
+use {
+    crate::{
+        persistence::Persistence,
+        types::{
+            errors::PersistenceError,
+            shared::{PartialEvent, PartialEventKey, StreamUpdate},
+        },
+    },
+    dashmap::DashMap,
+    solana_sdk::pubkey::Pubkey,
+    std::sync::Arc,
+    tokio::sync::mpsc::Receiver,
+};
+
+/// Decoder component.
+///
+/// The watchdog holds a clone of the same `partials` map, so the two operate on
+/// the same concurrent map without any message passing between them.
+pub(crate) struct Decoder {
+    /// Persistence layer.
+    pub persistence: Persistence,
+
+    /// Incoming `StreamUpdate` from the ingester.
+    pub rx: Receiver<StreamUpdate>,
+
+    /// Shared in-memory map of partial events keyed by `PartialEventKey`,
+    /// holding either-half events waiting for their pair. The watchdog holds a
+    /// clone of this `Arc`.
+    pub partials: Arc<DashMap<PartialEventKey, PartialEvent>>,
+
+    /// Settlement program id (filter target for the decoder).
+    pub settlement_program: Pubkey,
+
+    /// SolFlow program id (filter target for the decoder).
+    pub solflow_program: Pubkey,
+}
+
+impl Decoder {
+    /// Construct a new decoder. The caller owns the channel capacity decision.
+    pub fn new(
+        persistence: Persistence,
+        rx: Receiver<StreamUpdate>,
+        partials: Arc<DashMap<PartialEventKey, PartialEvent>>,
+        settlement_program: Pubkey,
+        solflow_program: Pubkey,
+    ) -> Self {
+        Self {
+            persistence,
+            rx,
+            partials,
+            settlement_program,
+            solflow_program,
+        }
+    }
+
+    /// Main loop. Pulls `StreamUpdate` from the receiver, runs the decode
+    /// pipeline, persists, and records partial events in the shared map for the
+    /// watchdog to read.
+    pub async fn run(&mut self) -> Result<(), PersistenceError> {
+        unimplemented!()
+    }
+}

crates/solana-indexer/src/indexer/finalization.rs

@@ -0,0 +1,66 @@
+#![expect(dead_code)]
+//! The finalization worker updates the commitment level of the transactions
+//! tracked by the indexer, promoting rows written at `confirmed` to
+//! `finalized`.
+//!
+//! It does so through two flows. Two are needed because the relevant RPC
+//! methods trade off differently: `getSignatureStatuses` is batchable but the
+//! node only retains statuses for recent slots, while `getTransaction` reaches
+//! arbitrarily old transactions on archival nodes but costs one call per
+//! signature. The batched pass handles the common case cheaply; the per-row
+//! sweep catches rows that age out of it.
+//!
+//! - **Promotion pass**: batch-polls `getSignatureStatuses` (at most
+//!   [`PROMOTION_BATCH_LIMIT`] signatures per call) over rows still at
+//!   `confirmed` that are at least [`FINALIZATION_WINDOW_SLOTS`] behind the
+//!   chain tip, and promotes rows whose `confirmationStatus` is `"finalized"`.
+//!
+//! - **Aged-row sweep**: fallback for rows past the signature-status retention
+//!   horizon ([`SIGNATURE_STATUS_RETENTION_SLOTS`]), which the promotion pass
+//!   can no longer check. Each row costs one `getTransaction` call; a non-null
+//!   response promotes to `finalized`, a null response marks `rolled_back`.
+
+// TODO:  This file only declares the component skeleton. The `run` body is
+// `unimplemented!`; both flows arrive in a later change.
+
+use {
+    crate::{persistence::Persistence, traits::solana_client::SolanaClient},
+    std::sync::Arc,
+};
+
+/// Slots a transaction usually needs to finalize (~12.8 s at 400 ms/slot).
+/// A heuristic floor, not a guarantee: the promotion pass skips rows fresher
+/// than this because they cannot have finalized yet, and degraded consensus
+/// can push real finalization later (the aged-row sweep catches those).
+pub const FINALIZATION_WINDOW_SLOTS: u64 = 32;
+
+/// Upper limit for the `getSignatureStatuses` batch RPC call.
+pub const PROMOTION_BATCH_LIMIT: usize = 256;
+
+/// Approximate slot horizon past which `getSignatureStatuses` no longer returns
+/// a result.
+pub const SIGNATURE_STATUS_RETENTION_SLOTS: u64 = 150;
+
+/// Transaction finalization worker. See the module docs for the two flows it
+/// runs.
+pub(crate) struct FinalizationWorker {
+    /// Persistence layer.
+    pub persistence: Persistence,
+
+    /// RPC implementor.
+    pub rpc: Arc<dyn SolanaClient>,
+}
+
+impl FinalizationWorker {
+    /// Construct a new finalization worker.
+    pub fn new(persistence: Persistence, rpc: Arc<dyn SolanaClient>) -> Self {
+        Self { persistence, rpc }
+    }
+
+    /// Outer loop. Runs the promotion pass and the aged-row sweep on a timer.
+    ///
+    /// Placeholder for now; implemented in a later change.
+    pub async fn run(&mut self) {
+        unimplemented!("implemented in PR 11–12")
+    }
+}

crates/solana-indexer/src/indexer/ingester.rs

@@ -0,0 +1,63 @@
+#![expect(dead_code)]
+//! The ingester owns the yellowstone gRPC stream. It drains the socket as fast
+//! as yellowstone delivers, pushes tagged updates into the channel, and updates
+//! `LATEST_CHAIN_SLOT` on every slot-filter message. It performs no decoding.
+
+//  TODO: This file only declares the component skeleton. The `run` body is
+// `unimplemented!`; the actual drain and reconnect with backoff logic arrives
+// in a later change.
+
+use {
+    crate::{persistence::Persistence, types::shared::StreamUpdate},
+    std::sync::atomic::AtomicU64,
+    tokio::sync::mpsc::Sender,
+    yellowstone_grpc_client::GrpcConnector,
+};
+
+/// The sole writer is the ingester, on every slot-filter message. Anchors the
+/// partial-event watchdog and the finalization worker. Cold start is zero; the
+/// watchdog skips its comparison on the first tick.
+///
+/// This is the chain tip, not indexing progress. How far the indexer has
+/// actually persisted is the watermark in `solana.indexer_state`, written by
+/// the decoder, which is a separate value.
+pub static LATEST_CHAIN_SLOT: AtomicU64 = AtomicU64::new(0);
+
+/// Cap on the exponential backoff between reconnect attempts.
+pub const RECONNECT_BACKOFF_CAP: std::time::Duration = std::time::Duration::from_secs(30);
+
+/// Capacity of the channel from the ingester to the decoder.
+pub const INGEST_TO_DECODER_CAPACITY: usize = 1024;
+
+/// Ingester component.
+///
+/// Generic over a `GrpcConnector` implementor so the unit tests can drive it
+/// with a mock.
+pub(crate) struct Ingester<C: GrpcConnector> {
+    /// gRPC connector implementor
+    pub connector: C,
+
+    /// Sends `StreamUpdate` to the decoder. Should be bounded to
+    /// `INGEST_TO_DECODER_CAPACITY` entries.
+    pub tx: Sender<StreamUpdate>,
+
+    /// Persistence layer, used to checkpoint the slot.
+    pub persistence: Persistence,
+}
+
+impl<C: GrpcConnector> Ingester<C> {
+    /// Construct a new ingester. The caller owns the channel capacity decision.
+    pub fn new(connector: C, tx: Sender<StreamUpdate>, persistence: Persistence) -> Self {
+        Self {
+            connector,
+            tx,
+            persistence,
+        }
+    }
+
+    /// TODO: Outer loop: open the subscription, drain it, push into the
+    /// channel, reconnect on failure with exponential backoff.
+    pub async fn run(&mut self) {
+        unimplemented!()
+    }
+}