diff --git a/crates/solana-indexer/src/indexer/decoder/tests.rs b/crates/solana-indexer/src/indexer/decoder/tests.rs index 8bfcce3b8e..241d4f2dde 100644 --- a/crates/solana-indexer/src/indexer/decoder/tests.rs +++ b/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) { 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), + }] + ); +} diff --git a/crates/solana-indexer/src/persistence.rs b/crates/solana-indexer/src/persistence.rs index 579c3bb723..2c55e49437 100644 --- a/crates/solana-indexer/src/persistence.rs +++ b/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, + 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` 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>>, +} + +#[cfg(test)] +impl Persistence { + /// The writes this instance received, in order. + pub(crate) fn calls(&self) -> Vec { + 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,6 +65,11 @@ 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(()) } @@ -39,6 +77,8 @@ impl Persistence { 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(()) }