feat(solana-indexer) PR 9: cover the ingester to decoder pipeline end to end

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

@@ -1,7 +1,8 @@
 use {
     super::{Decoder, ResolvedOrder, build_account_keys, decode_settlement, relevant_instructions},
     crate::{
-        persistence::Persistence,
+        indexer::ingester::Ingester,
+        persistence::{Call, Persistence},
         types::{
             Signature,
             channel::StreamUpdate,
@@ -14,20 +15,25 @@ use {
                 InnerInstruction,
                 InnerInstructions,
                 Message,
+                SubscribeUpdate,
+                SubscribeUpdateTransaction,
                 SubscribeUpdateTransactionInfo,
                 Transaction,
                 TransactionError,
                 TransactionStatusMeta,
+                UpdateOneof,
             },
         },
     },
     bytes::Bytes,
+    futures::stream,
     settlement_interface::{
         Pubkey as InterfacePubkey,
         SettlementInstruction,
         data::intent::{EncodedOrderIntent, OrderIntent, OrderKind},
     },
     solana_sdk::pubkey::Pubkey,
+    std::sync::{Arc, atomic::AtomicU64},
     tokio::sync::mpsc::Sender,
 };
 
@@ -249,10 +255,21 @@ fn signature(n: u8) -> Signature {
 
 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);
+    let decoder = Decoder::new(Persistence::default(), rx, settlement, solflow);
     (decoder, sender)
 }
 
+/// Wrap a transaction fixture in the proto envelope the ingester reads.
+fn tx_update(slot: u64, info: SubscribeUpdateTransactionInfo) -> SubscribeUpdate {
+    SubscribeUpdate {
+        update_oneof: Some(UpdateOneof::Transaction(SubscribeUpdateTransaction {
+            slot,
+            transaction: Some(info),
+        })),
+        ..Default::default()
+    }
+}
+
 /// A transaction carrying one settlement instruction, so draining it also
 /// routes into `decode_settlement`.
 fn stream_tx(slot: Slot, signature: Signature, settlement: Pubkey) -> StreamUpdate {
@@ -554,3 +571,45 @@ fn begin_and_finalize_settle_decode_to_settlement_finalized() {
         }]
     );
 }
+
+/// Both components as one pipeline: a proto `SubscribeUpdate` carrying a
+/// `CreateOrder` goes into the ingester and comes out of the decoder as a
+/// persisted event. This is the only test spanning the channel, so it pins that
+/// what the ingester forwards is what the decoder can consume, and that the
+/// watermark riding along is the slot before the transaction's own (slot 42
+/// persists at 41).
+#[tokio::test]
+async fn ingester_to_decoder_persists_decoded_events() {
+    let (settlement, solflow) = (pubkey(1), pubkey(2));
+    let (mut info, expected_uid, created_by) = create_order_tx();
+    // The ingester drops transactions without a well-formed signature.
+    info.signature = signature(9).as_ref().to_vec();
+
+    let (sender, receiver) = tokio::sync::mpsc::channel(16);
+    let persistence = Persistence::default();
+    let mut ingester = Ingester::new(
+        stream::iter(vec![Ok(tx_update(42, info))]),
+        sender,
+        Arc::new(AtomicU64::new(0)),
+    );
+    let mut decoder = Decoder::new(persistence.clone(), receiver, settlement, solflow);
+
+    // Drive the ingester to the canned stream's end, which it reports as an
+    // error. `clean_stream_end_returns_stream_ended` pins which one.
+    ingester.run().await.unwrap_err();
+    // Dropping the ingester closes the channel so the decoder's drain returns.
+    drop(ingester);
+    assert!(decoder.run().await.is_ok());
+
+    assert_eq!(
+        persistence.calls(),
+        vec![Call::PersistEvents {
+            events: vec![DecodedEvent::Settlement(SettlementEvent::OrderCreated {
+                order_uid: expected_uid,
+                owner: Pubkey::new_from_array([0x11; 32]),
+                created_by,
+            })],
+            watermark: Slot(41),
+        }]
+    );
+}

crates/solana-indexer/src/persistence.rs

@@ -13,13 +13,46 @@ use {
     std::ops::RangeInclusive,
 };
 
+/// One write the decoder asked for, captured so tests can assert the persist
+/// contract without a database behind it.
+#[cfg(test)]
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub(crate) enum Call {
+    /// Events plus the watermark they ride with.
+    PersistEvents {
+        events: Vec<DecodedEvent>,
+        watermark: Slot,
+    },
+    /// A watermark advance on a transaction that decoded to no events.
+    Watermark(Slot),
+    /// A transaction whose decode failed.
+    DeadLetter { signature: Signature, slot: Slot },
+}
+
 /// PostgreSQL persistence. Used by Decoder, Watchdog, and FinalizationWorker.
 ///
 /// Cheap to clone: wraps a shared pool. The method bodies are stubbed until the
 /// Postgres adapter lands.
 // TODO: hold `postgres: Arc<Postgres>` once the adapter is added.
-#[derive(Clone)]
-pub(crate) struct Persistence {}
+#[derive(Clone, Default)]
+pub(crate) struct Persistence {
+    /// Shared with every clone, so a test can read what the decoder wrote after
+    /// handing its own clone to the component.
+    #[cfg(test)]
+    calls: std::sync::Arc<std::sync::Mutex<Vec<Call>>>,
+}
+
+#[cfg(test)]
+impl Persistence {
+    /// The writes this instance received, in order.
+    pub(crate) fn calls(&self) -> Vec<Call> {
+        self.calls.lock().unwrap().clone()
+    }
+
+    fn record(&self, call: Call) {
+        self.calls.lock().unwrap().push(call);
+    }
+}
 
 impl Persistence {
     /// Save decoded events and advance the slot watermark atomically.
@@ -32,13 +65,20 @@ impl Persistence {
         // events and advances the watermark in one SQL transaction: append rows
         // as INSERT ON CONFLICT DO NOTHING, the watermark UPDATE guarded with
         // WHERE slot < $new_watermark.
+        #[cfg(test)]
+        self.record(Call::PersistEvents {
+            events,
+            watermark: new_watermark,
+        });
         Ok(())
     }
 
     /// Record a slot checkpoint. Rejects downward writes.
     pub(crate) async fn write_watermark(&self, slot: Slot) -> Result<(), PersistenceError> {
         // No-op seam until the Postgres adapter lands, which adds the monotonic
         // guard.
+        #[cfg(test)]
+        self.record(Call::Watermark(slot));
         Ok(())
     }
 
@@ -55,6 +95,8 @@ impl Persistence {
         slot: Slot,
     ) -> Result<(), PersistenceError> {
         // No-op seam until the Postgres adapter lands.
+        #[cfg(test)]
+        self.record(Call::DeadLetter { signature, slot });
         Ok(())
     }