From 5a0fbec8bfff667e1981c8a2af0afdd73401faf6 Mon Sep 17 00:00:00 2001 From: squadgazzz Date: Tue, 28 Jul 2026 12:25:16 +0000 Subject: [PATCH 1/4] test(solana-indexer): cover the ingester to decoder pipeline end to end --- .../src/indexer/decoder/tests.rs | 65 ++++++++++++++++++- crates/solana-indexer/src/persistence.rs | 40 +++++++++++- 2 files changed, 101 insertions(+), 4 deletions(-) diff --git a/crates/solana-indexer/src/indexer/decoder/tests.rs b/crates/solana-indexer/src/indexer/decoder/tests.rs index 8bfcce3b8e..7a90460436 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::{Error as IngesterError, 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,47 @@ 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); + + // The ingester forwards the update, then reports the canned stream's end. + assert!(matches!( + ingester.run().await, + Err(IngesterError::StreamEnded) + )); + // 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( + vec![DecodedEvent::Settlement(SettlementEvent::OrderCreated { + order_uid: expected_uid, + owner: Pubkey::new_from_array([0x11; 32]), + created_by, + })], + 41, + )] + ); +} diff --git a/crates/solana-indexer/src/persistence.rs b/crates/solana-indexer/src/persistence.rs index 30b322ff50..4f01d0e9d3 100644 --- a/crates/solana-indexer/src/persistence.rs +++ b/crates/solana-indexer/src/persistence.rs @@ -13,13 +13,43 @@ 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(Vec, u64), + /// A watermark advance on a transaction that decoded to no events. + Watermark(u64), + /// A transaction whose decode failed. + DeadLetter(Signature, Slot, &'static str), +} + /// 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 +62,8 @@ 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, new_watermark)); Ok(()) } @@ -39,6 +71,8 @@ impl Persistence { pub(crate) async fn write_watermark(&self, slot: u64) -> Result<(), PersistenceError> { // No-op seam until the Postgres adapter lands, which adds the monotonic // guard. + #[cfg(test)] + self.record(Call::Watermark(slot)); Ok(()) } @@ -52,6 +86,8 @@ impl Persistence { reason: &'static str, ) -> Result<(), PersistenceError> { // No-op seam until the Postgres adapter lands. + #[cfg(test)] + self.record(Call::DeadLetter(signature, slot, reason)); Ok(()) } From 0e2e3c4f9ecece99dc07596bc7be39586062fdc7 Mon Sep 17 00:00:00 2001 From: squadgazzz Date: Tue, 28 Jul 2026 12:49:09 +0000 Subject: [PATCH 2/4] test(solana-indexer): drop the redundant stream-end variant assertion --- crates/solana-indexer/src/indexer/decoder/tests.rs | 10 ++++------ 1 file changed, 4 insertions(+), 6 deletions(-) diff --git a/crates/solana-indexer/src/indexer/decoder/tests.rs b/crates/solana-indexer/src/indexer/decoder/tests.rs index 7a90460436..99dfbbcdcb 100644 --- a/crates/solana-indexer/src/indexer/decoder/tests.rs +++ b/crates/solana-indexer/src/indexer/decoder/tests.rs @@ -1,7 +1,7 @@ use { super::{Decoder, ResolvedOrder, build_account_keys, decode_settlement, relevant_instructions}, crate::{ - indexer::ingester::{Error as IngesterError, Ingester}, + indexer::ingester::Ingester, persistence::{Call, Persistence}, types::{ Signature, @@ -594,11 +594,9 @@ async fn ingester_to_decoder_persists_decoded_events() { ); let mut decoder = Decoder::new(persistence.clone(), receiver, settlement, solflow); - // The ingester forwards the update, then reports the canned stream's end. - assert!(matches!( - ingester.run().await, - Err(IngesterError::StreamEnded) - )); + // 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()); From 4b11c869394fa200dc595a30b7bca802ead6de56 Mon Sep 17 00:00:00 2001 From: squadgazzz Date: Tue, 28 Jul 2026 13:02:51 +0000 Subject: [PATCH 3/4] refactor(solana-indexer): type the persistence slot arguments as Slot --- crates/solana-indexer/src/indexer/decoder.rs | 2 +- crates/solana-indexer/src/indexer/decoder/tests.rs | 2 +- crates/solana-indexer/src/indexer/ingester.rs | 4 +++- crates/solana-indexer/src/persistence.rs | 14 +++++++------- 4 files changed, 12 insertions(+), 10 deletions(-) diff --git a/crates/solana-indexer/src/indexer/decoder.rs b/crates/solana-indexer/src/indexer/decoder.rs index 1fefa3ba70..7dda2effe5 100644 --- a/crates/solana-indexer/src/indexer/decoder.rs +++ b/crates/solana-indexer/src/indexer/decoder.rs @@ -86,7 +86,7 @@ impl Decoder { // 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 = u64::from(slot).saturating_sub(1); + let watermark = Slot(u64::from(slot).saturating_sub(1)); if events.is_empty() { self.persistence.write_watermark(watermark).await?; } else { diff --git a/crates/solana-indexer/src/indexer/decoder/tests.rs b/crates/solana-indexer/src/indexer/decoder/tests.rs index 99dfbbcdcb..d2f0d293e1 100644 --- a/crates/solana-indexer/src/indexer/decoder/tests.rs +++ b/crates/solana-indexer/src/indexer/decoder/tests.rs @@ -609,7 +609,7 @@ async fn ingester_to_decoder_persists_decoded_events() { owner: Pubkey::new_from_array([0x11; 32]), created_by, })], - 41, + Slot(41), )] ); } diff --git a/crates/solana-indexer/src/indexer/ingester.rs b/crates/solana-indexer/src/indexer/ingester.rs index a2253dbbbb..1fe1e26929 100644 --- a/crates/solana-indexer/src/indexer/ingester.rs +++ b/crates/solana-indexer/src/indexer/ingester.rs @@ -266,10 +266,12 @@ impl Ingester { settlement_program: Pubkey, solflow_program: Pubkey, ) -> Result<(), Error> { + // The proto field is a bare slot number, and `from_slot` is inclusive, so + // resume one past the last fully persisted slot. let from_slot = persistence .read_watermark() .await? - .map(|watermark| watermark + 1); + .map(|watermark| u64::from(watermark) + 1); let request = subscribe_request(settlement_program, solflow_program, from_slot); // The sink is the bidi request half: if kept, it can reconfigure the diff --git a/crates/solana-indexer/src/persistence.rs b/crates/solana-indexer/src/persistence.rs index 4f01d0e9d3..34ac6f536f 100644 --- a/crates/solana-indexer/src/persistence.rs +++ b/crates/solana-indexer/src/persistence.rs @@ -19,9 +19,9 @@ use { #[derive(Debug, Clone, PartialEq, Eq)] pub(crate) enum Call { /// Events plus the watermark they ride with. - PersistEvents(Vec, u64), + PersistEvents(Vec, Slot), /// A watermark advance on a transaction that decoded to no events. - Watermark(u64), + Watermark(Slot), /// A transaction whose decode failed. DeadLetter(Signature, Slot, &'static str), } @@ -56,7 +56,7 @@ impl Persistence { pub(crate) async fn persist_events( &self, events: Vec, - new_watermark: u64, + new_watermark: Slot, ) -> Result<(), PersistenceError> { // No-op seam until the Postgres adapter lands. The adapter writes the // events and advances the watermark in one SQL transaction: append rows @@ -68,7 +68,7 @@ impl Persistence { } /// Record a slot checkpoint. Rejects downward writes. - pub(crate) async fn write_watermark(&self, slot: u64) -> Result<(), PersistenceError> { + 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)] @@ -92,14 +92,14 @@ impl Persistence { } /// Read persisted watermark for resuming after reconnect. - pub(crate) async fn read_watermark(&self) -> Result, PersistenceError> { + pub(crate) async fn read_watermark(&self) -> Result, PersistenceError> { todo!() } /// Record gaps that fell outside the replay window (write-only in v0.1). pub(crate) async fn record_lost_slot_range( &self, - range: RangeInclusive, + range: RangeInclusive, ) -> Result<(), PersistenceError> { todo!() } @@ -126,7 +126,7 @@ impl Persistence { /// (see `get_confirmed_rows`). pub(crate) async fn get_aged_rows( &self, - retention_horizon_slot: u64, + retention_horizon_slot: Slot, ) -> Result, PersistenceError> { todo!() } From bada91cc1400a9614673795d4b496a35259482da Mon Sep 17 00:00:00 2001 From: squadgazzz Date: Tue, 28 Jul 2026 13:18:29 +0000 Subject: [PATCH 4/4] refactor(solana-indexer): name the recorded call fields and drop the dead-letter reason argument --- crates/solana-indexer/src/indexer/decoder.rs | 4 +--- .../src/indexer/decoder/tests.rs | 8 +++---- crates/solana-indexer/src/persistence.rs | 23 +++++++++++++------ 3 files changed, 21 insertions(+), 14 deletions(-) diff --git a/crates/solana-indexer/src/indexer/decoder.rs b/crates/solana-indexer/src/indexer/decoder.rs index 7dda2effe5..e409ff44a0 100644 --- a/crates/solana-indexer/src/indexer/decoder.rs +++ b/crates/solana-indexer/src/indexer/decoder.rs @@ -98,9 +98,7 @@ impl Decoder { // 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, "decoder_error") - .await?; + self.persistence.write_dead_letter(signature, slot).await?; } } Ok(()) diff --git a/crates/solana-indexer/src/indexer/decoder/tests.rs b/crates/solana-indexer/src/indexer/decoder/tests.rs index d2f0d293e1..241d4f2dde 100644 --- a/crates/solana-indexer/src/indexer/decoder/tests.rs +++ b/crates/solana-indexer/src/indexer/decoder/tests.rs @@ -603,13 +603,13 @@ async fn ingester_to_decoder_persists_decoded_events() { assert_eq!( persistence.calls(), - vec![Call::PersistEvents( - vec![DecodedEvent::Settlement(SettlementEvent::OrderCreated { + vec![Call::PersistEvents { + events: vec![DecodedEvent::Settlement(SettlementEvent::OrderCreated { order_uid: expected_uid, owner: Pubkey::new_from_array([0x11; 32]), created_by, })], - Slot(41), - )] + watermark: Slot(41), + }] ); } diff --git a/crates/solana-indexer/src/persistence.rs b/crates/solana-indexer/src/persistence.rs index 34ac6f536f..2c55e49437 100644 --- a/crates/solana-indexer/src/persistence.rs +++ b/crates/solana-indexer/src/persistence.rs @@ -19,11 +19,14 @@ use { #[derive(Debug, Clone, PartialEq, Eq)] pub(crate) enum Call { /// Events plus the watermark they ride with. - PersistEvents(Vec, Slot), + 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, Slot, &'static str), + DeadLetter { signature: Signature, slot: Slot }, } /// PostgreSQL persistence. Used by Decoder, Watchdog, and FinalizationWorker. @@ -63,7 +66,10 @@ impl Persistence { // as INSERT ON CONFLICT DO NOTHING, the watermark UPDATE guarded with // WHERE slot < $new_watermark. #[cfg(test)] - self.record(Call::PersistEvents(events, new_watermark)); + self.record(Call::PersistEvents { + events, + watermark: new_watermark, + }); Ok(()) } @@ -77,17 +83,20 @@ impl Persistence { } /// Record a transaction whose decode failed so recovery can replay it by - /// signature. One row per transaction. `reason` is always "decoder_error" - /// at v0.1. + /// signature. One row per transaction. + /// + /// The row's `reason` column is not a parameter: a decoder error is the + /// only failure mode that reaches this table (spec ยง12.2), so the + /// adapter writes `'decoder_error'`. A second reason would arrive as a + /// typed argument. pub(crate) async fn write_dead_letter( &self, signature: Signature, slot: Slot, - reason: &'static str, ) -> Result<(), PersistenceError> { // No-op seam until the Postgres adapter lands. #[cfg(test)] - self.record(Call::DeadLetter(signature, slot, reason)); + self.record(Call::DeadLetter { signature, slot }); Ok(()) }