feat(solana-indexer) PR 7.1: wire tx-only decoder run loop

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

@@ -2,8 +2,9 @@
 //! The decoder pulls `StreamUpdate`s from the ingester, decodes
 //! settlement-program and SolFlow transactions, and persists typed events.
 
-// TODO: `run` is unimplemented. The dispatch and persist steps that consume the
-// resolved instructions below are not wired up yet.
+// TODO: `decode_settlement`/`decode_solflow` and the persist path are stubbed.
+// `run` drains the channel and routes each transaction's tracked instructions
+// to those stubs.
 
 use {
     crate::{
@@ -51,10 +52,46 @@ impl Decoder {
         }
     }
 
-    /// Main loop. Pulls `StreamUpdate` from the receiver, runs the decode
-    /// pipeline, and persists typed events.
+    /// Main loop. Drains the channel and routes each transaction's tracked
+    /// instructions to their per-program decoders. Returns when the ingester
+    /// drops the sender.
     pub async fn run(&mut self) -> Result<(), PersistenceError> {
-        unimplemented!()
+        while let Some(update) = self.rx.recv().await {
+            let StreamUpdate::Tx { inner, .. } = update;
+            self.decode(&inner);
+        }
+        Ok(())
+    }
+
+    /// Route each tracked instruction in the transaction to its per-program
+    /// decoder.
+    fn decode(&self, tx: &SubscribeUpdateTransactionInfo) {
+        for instruction in
+            relevant_instructions(tx, &self.settlement_program, &self.solflow_program)
+        {
+            if instruction.program_id == self.settlement_program {
+                self.decode_settlement(&instruction);
+            } else {
+                self.decode_solflow(&instruction);
+            }
+        }
+    }
+
+    /// TODO: decode the settlement instruction data into typed events.
+    fn decode_settlement(&self, instruction: &ResolvedInstruction) {
+        tracing::debug!(
+            instruction_index = instruction.instruction_index,
+            "settlement instruction decode not implemented"
+        );
+    }
+
+    /// TODO: decode the SolFlow instruction data. The on-chain program does not
+    /// exist yet.
+    fn decode_solflow(&self, instruction: &ResolvedInstruction) {
+        tracing::debug!(
+            instruction_index = instruction.instruction_index,
+            "sol_flow instruction decode not implemented"
+        );
     }
 }
 

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

@@ -1,16 +1,25 @@
 use {
-    super::{build_account_keys, relevant_instructions},
-    crate::types::wire::{
-        CompiledInstruction,
-        InnerInstruction,
-        InnerInstructions,
-        Message,
-        SubscribeUpdateTransactionInfo,
-        Transaction,
-        TransactionStatusMeta,
+    super::{Decoder, build_account_keys, relevant_instructions},
+    crate::{
+        persistence::Persistence,
+        types::{
+            Signature,
+            channel::StreamUpdate,
+            slot::Slot,
+            wire::{
+                CompiledInstruction,
+                InnerInstruction,
+                InnerInstructions,
+                Message,
+                SubscribeUpdateTransactionInfo,
+                Transaction,
+                TransactionStatusMeta,
+            },
+        },
     },
     bytes::Bytes,
     solana_sdk::pubkey::Pubkey,
+    tokio::sync::mpsc::Sender,
 };
 
 fn pubkey(n: u8) -> Pubkey {
@@ -224,3 +233,44 @@ fn corrupt_stack_height_is_clamped() {
     // depth 9999 clamped to 4, so the path is bounded, not 9999 elements
     assert_eq!(relevant[0].inner_ix_path, vec![0, 0, 0, 0]);
 }
+
+fn signature(n: u8) -> Signature {
+    Signature::from([n; 64])
+}
+
+fn test_decoder(settlement: Pubkey, solflow: Pubkey) -> (Decoder, Sender<StreamUpdate>) {
+    let (sender, rx) = tokio::sync::mpsc::channel(16);
+    let decoder = Decoder::new(Persistence {}, rx, settlement, solflow);
+    (decoder, sender)
+}
+
+/// A transaction carrying one settlement instruction, so draining it also
+/// routes into `decode_settlement`.
+fn stream_tx(slot: Slot, signature: Signature, settlement: Pubkey) -> StreamUpdate {
+    let info = tx_info(
+        vec![settlement, pubkey(8)],
+        vec![],
+        vec![],
+        vec![compiled(0, vec![1], vec![0])],
+        vec![],
+    );
+    StreamUpdate::Tx {
+        slot,
+        signature,
+        inner: Box::new(info),
+    }
+}
+
+#[tokio::test]
+async fn run_drains_transactions_until_the_sender_drops() {
+    let (settlement, solflow) = (pubkey(1), pubkey(2));
+    let (mut decoder, sender) = test_decoder(settlement, solflow);
+
+    sender
+        .send(stream_tx(Slot(7), signature(3), settlement))
+        .await
+        .unwrap();
+    drop(sender);
+
+    assert!(decoder.run().await.is_ok());
+}

crates/solana-indexer/src/types/channel.rs

@@ -1,4 +1,4 @@
-#![expect(dead_code)]
+#![allow(dead_code)]
 //! Message types passed over the internal channel.
 //!
 //! The ingester pushes [`StreamUpdate`] into the channel to the decoder.