solana-indexer: revert rolled-back transactions when their slots finalize

crates/autopilot-svm/src/infra/db.rs

@@ -43,6 +43,7 @@ SELECT o.uid, o.owner, o.sell_token, o.buy_token, o.sell_token_account,
 FROM solana.orders o
 LEFT JOIN solana.order_pda p ON p.order_uid = o.uid
 WHERE o.valid_to >= $1
+  AND NOT COALESCE(p.is_reorged, false)
   AND (o.valid_from IS NULL OR o.valid_from <= $1)
   AND (o.intent_signature IS NOT NULL
        OR o.presigned_transaction IS NOT NULL

crates/cow-solana-rpc/src/lib.rs

@@ -113,6 +113,25 @@ impl SolanaRPC {
         })
     }
 
+    /// Whether each signature exists on the node's chain. The lookup falls
+    /// back to the ledger history when a signature is past the recent-status
+    /// cache, so `false` means the transaction never landed or was rolled
+    /// back, not merely that it aged out. A status of any commitment level
+    /// counts as existing: the question here is presence, not finality.
+    pub async fn known_signatures(&self, signatures: &[Signature]) -> Result<Vec<bool>, Error> {
+        // getSignatureStatuses rejects calls with more than 256 signatures.
+        const CHUNK: usize = 256;
+        let mut known = Vec::with_capacity(signatures.len());
+        for chunk in signatures.chunks(CHUNK) {
+            let response = self
+                .inner
+                .get_signature_statuses_with_history(chunk)
+                .await?;
+            known.extend(response.value.into_iter().map(|status| status.is_some()));
+        }
+        Ok(known)
+    }
+
     /// Simulate a versioned transaction without sending it. Returns the
     /// simulation result including logs and any error.
     pub async fn simulate_transaction(

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

@@ -38,6 +38,7 @@ use {
         },
         recover_discriminator,
     },
+    cow_solana_rpc::SolanaRPC,
     solana_sdk::pubkey::Pubkey,
     std::collections::BTreeMap,
     tokio::sync::mpsc::Receiver,
@@ -48,6 +49,9 @@ pub(crate) struct Decoder {
     /// Persistence layer.
     pub persistence: Postgres,
 
+    /// RPC client the finalization audit checks signatures against.
+    pub rpc: SolanaRPC,
+
     /// Incoming `StreamUpdate` from the ingester.
     pub rx: Receiver<StreamUpdate>,
 
@@ -63,12 +67,14 @@ impl Decoder {
     /// Construct a new decoder. The caller owns the channel capacity decision.
     pub fn new(
         persistence: Postgres,
+        rpc: SolanaRPC,
         rx: Receiver<StreamUpdate>,
         settlement_program: Pubkey,
         solflow_program: Option<Pubkey>,
     ) -> Self {
         Self {
             persistence,
+            rpc,
             rx,
             settlement_program,
             solflow_program,
@@ -92,6 +98,7 @@ impl Decoder {
         // In-memory mirror of the persisted watermark, spares redundant
         // writes and flags late transactions.
         let mut watermark: Option<Slot> = None;
+        let mut finalized_through: Option<Slot> = None;
         while let Some(update) = self.rx.recv().await {
             let (slot, signature, inner) = match update {
                 StreamUpdate::Tx {
@@ -105,7 +112,7 @@ impl Decoder {
                     continue;
                 }
                 StreamUpdate::Finalized { slot } => {
-                    self.persistence.write_finalized_slot(slot).await?;
+                    self.finalize(slot, &mut finalized_through).await?;
                     continue;
                 }
             };
@@ -145,6 +152,55 @@ impl Decoder {
         Ok(())
     }
 
+    /// Advance the finalized watermark to `slot`, first auditing the rows
+    /// that finalization would freeze: every transaction indexed in the
+    /// newly final range must still exist on chain, and one that vanished
+    /// was rolled back, so its rows are reverted in the same SQL transaction
+    /// that advances the watermark. An audit RPC failure leaves the
+    /// watermark untouched, the next finalized status retries the range.
+    async fn finalize(
+        &self,
+        slot: Slot,
+        finalized_through: &mut Option<Slot>,
+    ) -> Result<(), PersistenceError> {
+        let after = match *finalized_through {
+            Some(after) => after,
+            // First finalized status since boot: resume the audit from the
+            // persisted watermark, or from this very slot on a fresh
+            // database, where nothing older was indexed.
+            None => self.persistence.finalized_slot().await?.unwrap_or(slot),
+        };
+        if slot <= after {
+            *finalized_through = Some(after);
+            return Ok(());
+        }
+        let signatures = self.persistence.unfinalized_signatures(after, slot).await?;
+        let vanished = if signatures.is_empty() {
+            Vec::new()
+        } else {
+            match self.rpc.known_signatures(&signatures).await {
+                Ok(known) => signatures
+                    .into_iter()
+                    .zip(known)
+                    .filter_map(|(signature, known)| (!known).then_some(signature))
+                    .collect(),
+                Err(err) => {
+                    tracing::warn!(?err, "signature audit failed, finalization delayed");
+                    *finalized_through = Some(after);
+                    return Ok(());
+                }
+            }
+        };
+        if vanished.is_empty() {
+            self.persistence.write_finalized_slot(slot).await?;
+        } else {
+            tracing::warn!(?vanished, "rolled-back transactions reverted");
+            self.persistence.finalize_through(slot, &vanished).await?;
+        }
+        *finalized_through = Some(slot);
+        Ok(())
+    }
+
     /// Flush every buffer at or below the confirmed slot, then advance the
     /// watermark to it: the slot's transactions all arrived before its
     /// status, and quiet slots below it have nothing to wait for.

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

@@ -40,6 +40,7 @@ use {
         data::intent::{Flags, OrderIntent, OrderKind as IntentOrderKind},
         pda::order::find_order_pda,
     },
+    cow_solana_rpc::{Mocks, RpcRequest, SolanaRPC},
     futures::StreamExt,
     solana_sdk::pubkey::Pubkey,
     std::sync::{Arc, atomic::AtomicU64},
@@ -396,7 +397,30 @@ fn create_order_tx() -> (SubscribeUpdateTransactionInfo, CreatedOrder) {
 fn pure_decoder(settlement: Pubkey, solflow: Pubkey) -> Decoder {
     let pool = sqlx::PgPool::connect_lazy("postgresql://").unwrap();
     let (_sender, rx) = tokio::sync::mpsc::channel(1);
-    Decoder::new(Postgres::new(pool), rx, settlement, Some(solflow))
+    Decoder::new(
+        Postgres::new(pool),
+        SolanaRPC::new_mock_with_mocks(Default::default()),
+        rx,
+        settlement,
+        Some(solflow),
+    )
+}
+
+/// The finalization audit asks for two signatures (the dead letter and the
+/// healthy create), both still known to the chain.
+fn mock_rpc_with_signature_statuses() -> SolanaRPC {
+    let status = serde_json::json!({
+        "slot": 43u64,
+        "confirmations": null,
+        "err": null,
+        "status": { "Ok": null },
+        "confirmationStatus": "finalized",
+    });
+    let statuses = serde_json::json!({
+        "context": { "slot": 43u64, "apiVersion": "2.0.0" },
+        "value": [status.clone(), status],
+    });
+    SolanaRPC::new_mock_with_mocks(Mocks::from([(RpcRequest::GetSignatureStatuses, statuses)]))
 }
 
 /// `decode` wraps settlement events as `DecodedEvent::Settlement` for `run`
@@ -655,6 +679,7 @@ async fn solana_db_ingester_to_decoder_persists_decoded_events() {
     let mut ingester = Ingester::new(geyser_stream, sender, Arc::new(AtomicU64::new(0)));
     let mut decoder = Decoder::new(
         Postgres::new(pool.clone()),
+        mock_rpc_with_signature_statuses(),
         receiver,
         settlement,
         Some(solflow),