feat(solana-indexer) PR 8: wire the decoder to the persistence seam

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

@@ -1,10 +1,9 @@
-#![expect(dead_code)]
+#![allow(dead_code, reason = "dead in the lib build, exercised by tests")]
 //! The decoder pulls `StreamUpdate`s from the ingester, decodes
 //! settlement-program and SolFlow transactions, and persists typed events.
 
-// TODO: `decode_solflow` and the persist path are stubbed. `run` drains the
-// channel, decodes settlement instructions into `SettlementEvent`s, and drops
-// them until the persistence adapter lands.
+// TODO: `decode_solflow` is a stub (no on-chain SolFlow program yet), and the
+// `Persistence` write bodies are no-ops until the Postgres adapter lands.
 
 use {
     crate::{
@@ -13,7 +12,7 @@ use {
             Signature,
             channel::StreamUpdate,
             errors::{DecodeError, PersistenceError},
-            events::{SettlementEvent, TradeDelta},
+            events::{DecodedEvent, SettlementEvent, TradeDelta},
             order::OrderUid,
             slot::Slot,
             tx::{ResolvedInstruction, TxContext},
@@ -68,44 +67,87 @@ impl Decoder {
         }
     }
 
-    /// Main loop. Drains the channel and decodes each transaction's tracked
-    /// instructions. Returns when the ingester drops the sender.
+    /// Main loop. Drains the channel, decodes each transaction's tracked
+    /// instructions, and hands the results to the persistence seam. Returns
+    /// when the ingester drops the sender.
     pub async fn run(&mut self) -> Result<(), PersistenceError> {
         while let Some(update) = self.rx.recv().await {
             let StreamUpdate::Tx {
                 slot,
                 signature,
                 inner,
             } = update;
-            self.decode(&inner, slot, signature);
+            let (events, decode_failed) = self.decode(&inner, slot, signature);
+            tracing::debug!(slot = %slot, event_count = events.len(), "decoded events");
+
+            // Stream resume is slot-granular (`from_slot = watermark + 1`), and
+            // the slot may still have more transactions in flight, so marking it
+            // done here could skip its remaining transactions after a crash.
+            // Writing `slot - 1` on every transaction only ever marks fully
+            // delivered slots. A redelivery of this slot after a restart is
+            // absorbed by idempotent writes.
+            let watermark = Slot(u64::from(slot).saturating_sub(1));
+            if events.is_empty() {
+                self.persistence.write_watermark(watermark).await?;
+            } else {
+                // The watermark advance rides the same call as the events so the
+                // adapter can apply both in one SQL transaction.
+                self.persistence.persist_events(events, watermark).await?;
+            }
+            if decode_failed {
+                // Partial decode is expected: events that did decode persist and
+                // the watermark advances. Recovery replays the whole transaction
+                // by signature, and idempotent writes absorb the overlap.
+                self.persistence.write_dead_letter(signature, slot).await?;
+            }
         }
         Ok(())
     }
 
-    /// Decode one transaction's tracked instructions into domain events. The
-    /// settlement half runs through the pure [`decode_settlement`]; the SolFlow
-    /// half is still stubbed.
-    fn decode(&self, tx: &SubscribeUpdateTransactionInfo, slot: Slot, signature: Signature) {
+    /// Decode one transaction's tracked instructions into domain events, plus
+    /// a flag reporting whether any settlement instruction failed to decode.
+    /// The settlement half runs through the pure [`decode_settlement`], the
+    /// SolFlow half is a stub.
+    #[tracing::instrument(skip_all, fields(slot = %slot, signature = %signature))]
+    fn decode(
+        &self,
+        tx: &SubscribeUpdateTransactionInfo,
+        slot: Slot,
+        signature: Signature,
+    ) -> (Vec<DecodedEvent>, bool) {
+        // `meta` carries whether the transaction succeeded, so without it there is
+        // no way to tell, and emitting events for a transaction that may have
+        // reverted is the failure this guard exists to prevent. Dead-letter it
+        // instead of skipping: replay re-fetches by signature, and
+        // `getTransaction` returns the meta.
+        let Some(meta) = tx.meta.as_ref() else {
+            tracing::warn!("transaction update without meta");
+            return (Vec::new(), true);
+        };
+
+        // A failed Solana transaction rolls back every account write, so no
+        // state-changing event may be emitted for it. Failed transactions are
+        // delivered on purpose (the revert is an attribution signal), and a
+        // dedicated revert-attribution event is a later PR.
+        if meta.err.is_some() {
+            tracing::debug!("transaction reverted, skipping");
+            return (Vec::new(), false);
+        }
+
         let instructions =
             relevant_instructions(tx, &self.settlement_program, &self.solflow_program);
         if instructions.is_empty() {
-            return;
+            return (Vec::new(), false);
         }
 
         // `relevant_instructions` reconstructs the account list internally to
         // resolve program ids; rebuild it once here so the decode can resolve
         // account indices to pubkeys too.
-        let account_keys = build_account_keys(tx);
-        let post_token_balances = tx
-            .meta
-            .as_ref()
-            .map(|meta| meta.post_token_balances.clone())
-            .unwrap_or_default();
         let ctx = TxContext {
             slot,
             signature,
-            account_keys,
-            post_token_balances,
+            account_keys: build_account_keys(tx),
+            post_token_balances: meta.post_token_balances.clone(),
         };
 
         // `relevant_instructions` yields only settlement and SolFlow instructions.
@@ -120,10 +162,7 @@ impl Decoder {
         // TODO: resolve order PDAs against persisted order rows once the store
         // adapter lands. Until then nothing resolves, so `SettlementFinalized`
         // events carry the tx-level fields with empty trades.
-        // TODO: skip transactions whose `meta.err` is set: a reverted settlement
-        // or order creation must not emit an event. Deferred until the persist
-        // path is wired (nothing is persisted yet).
-        let events = decode_settlement(&settlement, &ctx, |_order_pda| None);
+        let (events, decode_failed) = decode_settlement(&settlement, &ctx, |_order_pda| None);
 
         for instruction in instructions
             .iter()
@@ -132,13 +171,10 @@ impl Decoder {
             self.decode_solflow(instruction);
         }
 
-        // TODO: persist `events` once the persistence adapter lands; for now the
-        // decode runs end to end but its output is dropped.
-        tracing::debug!(
-            slot = %ctx.slot,
-            event_count = events.len(),
-            "decoded settlement events"
-        );
+        (
+            events.into_iter().map(DecodedEvent::Settlement).collect(),
+            decode_failed,
+        )
     }
 
     /// TODO: decode the SolFlow instruction data. The on-chain program does not
@@ -163,7 +199,8 @@ struct ResolvedOrder {
 }
 
 /// Decode the settlement-program instructions of one transaction into domain
-/// events.
+/// events. The `bool` is true when at least one instruction failed to decode,
+/// so the caller can dead-letter the transaction for replay.
 ///
 /// Pure: every tx-level input arrives through `ctx`, and any order field the
 /// `BeginSettle` wire does not carry is resolved through `resolve_order` (keyed
@@ -173,23 +210,27 @@ fn decode_settlement(
     instructions: &[ResolvedInstruction],
     ctx: &TxContext,
     resolve_order: impl Fn(&Pubkey) -> Option<ResolvedOrder>,
-) -> Vec<SettlementEvent> {
+) -> (Vec<SettlementEvent>, bool) {
     let mut events = Vec::new();
+    let mut decode_failed = false;
 
     // Instructions decodable on their own, without tx-level pairing.
     for instruction in instructions {
         let Ok((discriminator, _)) = recover_discriminator(&instruction.data) else {
-            tracing::debug!(
+            decode_failed = true;
+            // Warn, not debug: this dead-letters the transaction, so it needs to
+            // be findable in the logs alongside the row.
+            tracing::warn!(
                 instruction_index = instruction.instruction_index,
+                err = %DecodeError::UnknownDiscriminator,
                 "settlement instruction with an unknown discriminator, skipping"
             );
             continue;
         };
         // A landed (non-reverted) transaction carries valid instruction data, so
         // a decode failure here means a decoder bug or an unannounced program
-        // layout change, not a normal case. Surface it as a warning rather than
-        // dropping it silently. Once the persistence adapter lands these route to
-        // the dead-letter table.
+        // layout change, not a normal case. Surface it as a warning and set the
+        // failure flag so the transaction is dead-lettered.
         let decoded = match discriminator {
             SettlementInstruction::CreateOrder => {
                 decode_order_created(instruction, &ctx.account_keys).map(|event| vec![event])
@@ -208,11 +249,14 @@ fn decode_settlement(
         };
         match decoded {
             Ok(decoded_events) => events.extend(decoded_events),
-            Err(err) => tracing::warn!(
-                instruction_index = instruction.instruction_index,
-                %err,
-                "failed to decode settlement instruction"
-            ),
+            Err(err) => {
+                decode_failed = true;
+                tracing::warn!(
+                    instruction_index = instruction.instruction_index,
+                    %err,
+                    "failed to decode settlement instruction"
+                );
+            }
         }
     }
 
@@ -222,8 +266,9 @@ fn decode_settlement(
         instructions,
         ctx,
         &resolve_order,
+        &mut decode_failed,
     ));
-    events
+    (events, decode_failed)
 }
 
 /// `CreateOrder` -> `OrderCreated`. The parser recovers the encoded order
@@ -235,9 +280,9 @@ fn decode_order_created(
 ) -> Result<SettlementEvent, DecodeError> {
     let mut accounts = instruction_account_keys(instruction, account_keys)?;
     let input = CreateOrderInput::parse(&instruction.data, &mut accounts)
-        .map_err(|_| DecodeError::SchemaMismatch)?;
+        .map_err(|err| DecodeError::SchemaMismatch(err.to_string()))?;
     let (intent, uid) = EncodedOrderIntent::decode_and_hash(&input.intent_bytes)
-        .map_err(|_| DecodeError::SchemaMismatch)?;
+        .map_err(|err| DecodeError::SchemaMismatch(err.to_string()))?;
     Ok(SettlementEvent::OrderCreated {
         order_uid: OrderUid(uid.to_bytes()),
         owner: to_sdk_pubkey(intent.owner),
@@ -254,7 +299,7 @@ fn decode_buffers_created(
 ) -> Result<Vec<SettlementEvent>, DecodeError> {
     let mut accounts = instruction_account_keys(instruction, account_keys)?;
     let input = CreateBufferInput::parse(&instruction.data, &mut accounts)
-        .map_err(|_| DecodeError::SchemaMismatch)?;
+        .map_err(|err| DecodeError::SchemaMismatch(err.to_string()))?;
     Ok(input
         .buffers
         .iter()
@@ -268,11 +313,14 @@ fn decode_buffers_created(
 /// Pairing is by index: a parsed `BeginSettle` carries the top-level
 /// instruction index of its `FinalizeSettle` (`finalize_ix_index`), which must
 /// match a `FinalizeSettle` present in the same transaction. It is independent
-/// of the two instructions' relative order.
+/// of the two instructions' relative order. A pair that cannot be decoded (a
+/// parse failure, unresolvable account keys, or a `BeginSettle` whose named
+/// `FinalizeSettle` is missing) sets `decode_failed` and emits nothing.
 fn decode_settlements_finalized(
     instructions: &[ResolvedInstruction],
     ctx: &TxContext,
     resolve_order: &impl Fn(&Pubkey) -> Option<ResolvedOrder>,
+    decode_failed: &mut bool,
 ) -> Vec<SettlementEvent> {
     // The solver is the transaction fee payer: the first account key, which
     // Solana guarantees is the signer that submitted the transaction.
@@ -287,13 +335,23 @@ fn decode_settlements_finalized(
         };
         let mut begin_accounts = match instruction_account_keys(begin, &ctx.account_keys) {
             Ok(accounts) => accounts,
-            Err(_) => continue,
+            Err(err) => {
+                *decode_failed = true;
+                tracing::warn!(
+                    instruction_index = begin.instruction_index,
+                    %err,
+                    "BeginSettle account resolution failed, skipping"
+                );
+                continue;
+            }
         };
         let begin_input = match BeginSettleInput::parse(&begin.data, &mut begin_accounts) {
             Ok(input) => input,
-            Err(_) => {
+            Err(err) => {
+                *decode_failed = true;
                 tracing::warn!(
                     instruction_index = begin.instruction_index,
+                    %err,
                     "BeginSettle did not match the expected layout, skipping"
                 );
                 continue;
@@ -308,6 +366,7 @@ fn decode_settlements_finalized(
                     Ok((SettlementInstruction::FinalizeSettle, _))
                 )
         }) else {
+            *decode_failed = true;
             tracing::debug!(
                 instruction_index = begin.instruction_index,
                 finalize_ix_index = begin_input.finalize_ix_index,
@@ -317,14 +376,24 @@ fn decode_settlements_finalized(
         };
         let mut finalize_accounts = match instruction_account_keys(finalize, &ctx.account_keys) {
             Ok(accounts) => accounts,
-            Err(_) => continue,
+            Err(err) => {
+                *decode_failed = true;
+                tracing::warn!(
+                    instruction_index = finalize.instruction_index,
+                    %err,
+                    "FinalizeSettle account resolution failed, skipping"
+                );
+                continue;
+            }
         };
         let finalize_input =
             match FinalizeSettleInput::parse(&finalize.data, &mut finalize_accounts) {
                 Ok(input) => input,
-                Err(_) => {
+                Err(err) => {
+                    *decode_failed = true;
                     tracing::warn!(
                         instruction_index = finalize.instruction_index,
+                        %err,
                         "FinalizeSettle did not match the expected layout, skipping"
                     );
                     continue;
@@ -363,9 +432,7 @@ fn decode_settlements_finalized(
             .collect();
 
         events.push(SettlementEvent::SettlementFinalized {
-            // The wire carries `auction_id` as i64; it is non-negative in
-            // practice.
-            auction_id: begin_input.auction_id as u64,
+            auction_id: begin_input.auction_id,
             solver,
             tx_signature: ctx.signature,
             slot: ctx.slot,
@@ -386,10 +453,12 @@ fn instruction_account_keys(
         .accounts
         .iter()
         .map(|&index| {
-            account_keys
-                .get(usize::from(index))
-                .copied()
-                .ok_or(DecodeError::SchemaMismatch)
+            account_keys.get(usize::from(index)).copied().ok_or(
+                DecodeError::AccountIndexOutOfRange {
+                    index,
+                    len: account_keys.len(),
+                },
+            )
         })
         .collect()
 }