carbon_core/
transaction.rs

1//! Transaction-shaped pipe wiring + per-transaction metadata.
2//!
3//! # Components
4//!
5//! - [`TransactionMetadata`] — context shared across every instruction in one
6//!   transaction (signature, fee payer, slot, full message and meta).
7//! - [`TransactionProcessorInputType<'a, T>`] — borrowed bundle handed to
8//!   processors registered via `Pipeline::transaction(...)`.
9//! - [`TransactionPipe`] / [`TransactionPipes`] — internal pipe that parses
10//!   each instruction through an `InstructionDecoderCollection` and routes the
11//!   whole transaction to one processor.
12//! - [`parse_instructions_flat`] — helper that maps a flat list of instructions
13//!   through a collection.
14
15use {
16    crate::{
17        collection::InstructionDecoderCollection, error::CarbonResult, filter::Filter,
18        instruction::InstructionMetadata, processor::Processor,
19    },
20    async_trait::async_trait,
21    core::convert::TryFrom,
22    solana_hash::Hash,
23    solana_instruction::Instruction,
24    solana_pubkey::Pubkey,
25    solana_signature::Signature,
26    std::sync::Arc,
27};
28/// Per-transaction context shared across all of its instructions.
29///
30/// Built from a `TransactionUpdate` via `TryFrom`. Wrapped in `Arc`
31/// inside the pipeline so child instructions cheaply reference it.
32#[derive(Debug, Clone, Default)]
33pub struct TransactionMetadata {
34    pub slot: u64,
35    pub signature: Signature,
36    pub fee_payer: Pubkey,
37    pub meta: solana_transaction_status::TransactionStatusMeta,
38    pub message: solana_message::VersionedMessage,
39    pub index: Option<u64>,
40    pub block_time: Option<i64>,
41    pub block_hash: Option<Hash>,
42}
43
44impl TryFrom<crate::datasource::TransactionUpdate> for TransactionMetadata {
45    type Error = crate::error::Error;
46
47    fn try_from(value: crate::datasource::TransactionUpdate) -> Result<Self, Self::Error> {
48        let accounts = value.transaction.message.static_account_keys();
49
50        Ok(TransactionMetadata {
51            slot: value.slot,
52            signature: value.signature,
53            fee_payer: *accounts
54                .first()
55                .ok_or(crate::error::Error::MissingFeePayer)?,
56            meta: value.meta.clone(),
57            message: value.transaction.message.clone(),
58            index: value.index,
59            block_time: value.block_time,
60            block_hash: value.block_hash,
61        })
62    }
63}
64
65/// Borrowed bundle delivered to processors registered via
66/// `Pipeline::transaction(...)`: shared transaction metadata plus the
67/// flat list of decoded instructions.
68#[derive(Debug)]
69pub struct TransactionProcessorInputType<'a, T> {
70    pub metadata: &'a Arc<TransactionMetadata>,
71    pub instructions: &'a [(InstructionMetadata, T)],
72}
73
74pub struct TransactionPipe<T: InstructionDecoderCollection, P> {
75    processor: P,
76    filters: Vec<Box<dyn Filter + 'static>>,
77    _phantom: std::marker::PhantomData<T>,
78}
79
80impl<T: InstructionDecoderCollection, P> TransactionPipe<T, P> {
81    pub fn new(processor: P, filters: Vec<Box<dyn Filter + 'static>>) -> Self {
82        Self {
83            processor,
84            filters,
85            _phantom: std::marker::PhantomData,
86        }
87    }
88}
89
90#[async_trait]
91pub trait TransactionPipes<'a>: Send + Sync {
92    async fn run(
93        &mut self,
94        transaction_metadata: Arc<TransactionMetadata>,
95        instructions: &[(InstructionMetadata, Instruction)],
96    ) -> CarbonResult<()>;
97
98    fn filters(&self) -> &[Box<dyn Filter + 'static>];
99}
100
101#[async_trait]
102impl<T, P> TransactionPipes<'_> for TransactionPipe<T, P>
103where
104    T: InstructionDecoderCollection + Sync + 'static,
105    P: for<'a> Processor<TransactionProcessorInputType<'a, T>> + Send + Sync + 'static,
106{
107    async fn run(
108        &mut self,
109        transaction_metadata: Arc<TransactionMetadata>,
110        instructions: &[(InstructionMetadata, Instruction)],
111    ) -> CarbonResult<()> {
112        let parsed_instructions = parse_instructions_flat::<T>(instructions);
113
114        let data = TransactionProcessorInputType {
115            metadata: &transaction_metadata,
116            instructions: &parsed_instructions,
117        };
118
119        self.processor.process(&data).await?;
120
121        Ok(())
122    }
123
124    fn filters(&self) -> &[Box<dyn Filter + 'static>] {
125        &self.filters
126    }
127}
128
129pub fn parse_instructions_flat<T: InstructionDecoderCollection>(
130    instructions: &[(InstructionMetadata, Instruction)],
131) -> Vec<(InstructionMetadata, T)> {
132    instructions
133        .iter()
134        .filter_map(|(metadata, instruction)| {
135            T::parse_instruction(instruction).map(|parsed| (metadata.clone(), parsed))
136        })
137        .collect()
138}
carbon_core/transaction.rs

carbon_core/
transaction.rs
