Inner instructions for a Transaction pipe

[biancarosca]

Hi, I want to get all instructions inside the transaction pipe, but I only receive the main instructions and not the inner instructions. Is this supported or am I missing something?

This is what I do currently:

#[async_trait]
impl Processor for PumpfunTransactionProcessor {
    type InputType = TransactionProcessorInputType<
        PumpfunInstructionWrapper
    >;

    async fn process(
        &mut self,
        input: Self::InputType,
        _metrics: Arc<MetricsCollection>,
    ) -> CarbonResult<()> {

#[derive(Clone, Debug, Hash, Serialize, Deserialize, PartialEq, Eq)]
pub struct PumpfunInstructionWrapper(pub PumpfunInstruction);

impl InstructionDecoderCollection for PumpfunInstructionWrapper {
    type InstructionType = PumpfunInstruction;

    fn parse_instruction(
        instruction: &solana_sdk::instruction::Instruction,
    ) -> Option<DecodedInstruction<Self>> {
        if instruction.program_id == PUMPFUN_PROGRAM_ID {
            println!("{:?}", instruction);
            PumpfunDecoder
                .decode_instruction(instruction)
                .map(|decoded| DecodedInstruction {
                    program_id: decoded.program_id,
                    data: Self(decoded.data),
                    accounts: decoded.accounts,
                })
        } else {
            None
        }
    }

    fn get_type(&self) -> Self::InstructionType {
        self.0.clone()
    }
}

[biancarosca]

@KellianDev Can you maybe help?

[KellianDev]

Hello @biancarosca, I would recommend that you use an InstructionPipe for your use case, since you seem to only care about individual instructions and perhaps some metadata about the transaction.

Example pipeline:

carbon_core::pipeline::Pipeline::builder()
        .datasource(helius_websocket)
        .instruction(PumpfunDecoder, PumpfunInstructionProcessor)
        .build()?
        .run()
        .await?;

Example processor:

pub struct PumpfunInstructionProcessor;

#[async_trait]
impl Processor for PumpfunInstructionProcessor {
    type InputType = InstructionProcessorInputType<PumpfunInstruction>;

    async fn process(
        &mut self,
        data: Self::InputType,
        _metrics: Arc<MetricsCollection>,
    ) -> CarbonResult<()> {
        let (metadata, instruction, _) = input;
        
        let pumpfun_instruction: PumpfunInstruction = instruction.data;

        println!("\nInstruction stack height: {}", metadata.stack_height);

        match pumpfun_instruction {
            PumpfunInstruction::CreateEvent(create_event) => {
                println!("\nNew token created: {:#?}", create_event);
            }
            PumpfunInstruction::TradeEvent(trade_event) => {
                if trade_event.sol_amount > 10 * LAMPORTS_PER_SOL {
                    println!("\nBig trade occured: {:#?}", trade_event);
                }
            }
            PumpfunInstruction::CompleteEvent(complete_event) => {
                println!("\nBonded: {:#?}", complete_event);
            }
            _ => {}
        };

        Ok(())
    }
}

If you actually need to use the transaction pipe for another reason, give me more context and I'll be able to help you also.

[biancarosca]

I would like to receive all instructions in the transaction pipe because I need them all grouped together by transaction. If that is possible, can you show me an example?

[KellianDev]

It is possible, here you go:

Example pipeline:

carbon_core::pipeline::Pipeline::builder()
        .datasource(helius_websocket)
        .transaction(TransactionProcessor, None)
        .build()?
        .run()
        .await?;

Example instruction decoder collection and processor:

instruction_decoder_collection!(
      MyInstructions, MyInstructionsType, Programs,
      Pumpfun => carbon_pumpfun_decoder::PumpfunDecoder => carbon_pumpfun_decoder::PumpfunInstruction,
);

pub struct TransactionProcessor;
#[async_trait]
impl carbon_core::processor::Processor for TransactionProcessor {
    type InputType = TransactionProcessorInputType<MyInstructions>;

    async fn process(
        &mut self,
        data: Self::InputType,
        metrics: Arc<MetricsCollection>,
    ) -> CarbonResult<()> {
        let (metadata, instructions, _) = input;
       
        // whatever logic you need

        Ok(())
    }
}

In the transaction pipe, decoded instructions are given in a non nested, ordered struct, including inner instructions. You can look at stack height still in each instruction's metadata.
We lack documentation/examples about the transaction pipe right now, we will add more soon.