diff --git a/crates/solana-indexer/src/indexer/decoder.rs b/crates/solana-indexer/src/indexer/decoder.rs index 5d94c31b91..60a081c2d6 100644 --- a/crates/solana-indexer/src/indexer/decoder.rs +++ b/crates/solana-indexer/src/indexer/decoder.rs @@ -108,6 +108,10 @@ impl Decoder { .await?; continue; } + StreamUpdate::Finalized { slot } => { + self.persistence.write_finalized_slot(slot).await?; + continue; + } }; self.flush_up_to(&mut pending, slot, &mut flushed_through) .await?; @@ -164,6 +168,15 @@ impl Decoder { self.flush_slot(slot, buffer, true).await?; *flushed_through = (*flushed_through).max(Some(slot)); } + // Everything at or below the cutoff is complete even when nothing was + // buffered: advance the watermark on quiet slots too, so the resume + // point tracks the stream. + if cutoff > 0 && *flushed_through < Some(Slot(cutoff)) { + self.persistence + .write_last_indexed_slot(Slot(cutoff)) + .await?; + *flushed_through = Some(Slot(cutoff)); + } Ok(()) } diff --git a/crates/solana-indexer/src/indexer/decoder/tests.rs b/crates/solana-indexer/src/indexer/decoder/tests.rs index 962734a8ac..6501810192 100644 --- a/crates/solana-indexer/src/indexer/decoder/tests.rs +++ b/crates/solana-indexer/src/indexer/decoder/tests.rs @@ -21,6 +21,7 @@ use { InnerInstruction, InnerInstructions, Message, + SlotStatus, SubscribeUpdate, SubscribeUpdateSlot, SubscribeUpdateTransaction, @@ -313,10 +314,11 @@ fn signature(n: u8) -> Signature { } /// A slot-status message in the proto envelope the ingester reads. -fn slot_status_update(slot: u64) -> SubscribeUpdate { +fn slot_status_update(slot: u64, status: SlotStatus) -> SubscribeUpdate { SubscribeUpdate { update_oneof: Some(UpdateOneof::Slot(SubscribeUpdateSlot { slot, + status: status as i32, ..Default::default() })), ..Default::default() @@ -713,21 +715,41 @@ async fn solana_db_ingester_to_decoder_persists_decoded_events() { geyser_tx.send(update).await.unwrap(); } // The hold-back keeps both slots buffered: the newest observed slot (43) - // is not two past either of them, so nothing may be persisted yet. + // is not two past either of them, so no events may be persisted yet. The + // watermark still advances to the quiet slots below the hold-back. let reader = Postgres::new(pool.clone()); tokio::time::sleep(std::time::Duration::from_millis(200)).await; - assert_eq!(reader.last_indexed_slot().await.unwrap(), None); + assert_eq!(reader.last_indexed_slot().await.unwrap(), Some(Slot(41))); + let events: i64 = sqlx::query_scalar("SELECT count(*) FROM solana.order_pda") + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(events, 0); // The slot-45 status moves the stream two past 43 and flushes both slots. - // Closing the channel ends the ingester (a terminal stream end) and the - // decoder drains cleanly behind it, so joining both tasks is the - // guarantee that every write below has landed. - geyser_tx.send(Ok(slot_status_update(45))).await.unwrap(); + // The finalized status advances the finalized watermark, and the quiet + // slot 50 carries no transactions yet still advances the last indexed + // slot to 48 (the hold-back behind it). Closing the channel ends the + // ingester (a terminal stream end) and the decoder drains cleanly behind + // it, so joining both tasks is the guarantee that every write below has + // landed. + for update in [ + slot_status_update(45, SlotStatus::SlotConfirmed), + slot_status_update(43, SlotStatus::SlotFinalized), + slot_status_update(50, SlotStatus::SlotConfirmed), + ] { + geyser_tx.send(Ok(update)).await.unwrap(); + } drop(geyser_tx); assert!(ingester_task.await.unwrap().is_err()); assert!(decoder_task.await.unwrap().is_ok()); - assert_eq!(reader.last_indexed_slot().await.unwrap(), Some(Slot(43))); + assert_eq!(reader.last_indexed_slot().await.unwrap(), Some(Slot(48))); + let finalized: i64 = sqlx::query_scalar("SELECT finalized_slot FROM solana.indexer_state") + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(finalized, 43); // Slot 42 held only the reverted transaction: no dead letter, no rows. // The slot-43 transaction with the unknown discriminator is dead-lettered diff --git a/crates/solana-indexer/src/indexer/ingester.rs b/crates/solana-indexer/src/indexer/ingester.rs index 75be42402e..5c1641a5af 100644 --- a/crates/solana-indexer/src/indexer/ingester.rs +++ b/crates/solana-indexer/src/indexer/ingester.rs @@ -1,6 +1,6 @@ //! The ingester drains the yellowstone gRPC stream as fast as it delivers, //! pushes tagged updates into the channel, and advances the latest-chain-slot -//! counter on every slot-filter message. It performs no decoding. +//! counter on every confirmed slot message. It performs no decoding. //! //! The stream it drains is an `AutoReconnect`-backed //! [`GeyserStream`](yellowstone_grpc_client::GeyserStream) from @@ -30,6 +30,7 @@ use { slot::Slot, wire::{ CommitmentLevel, + SlotStatus, SubscribeRequest, SubscribeRequestFilterSlots, SubscribeRequestFilterTransactions, @@ -197,21 +198,37 @@ where .await } - /// Consume a slot message: advance the in-memory chain-tip counter and - /// forward the slot to the decoder so it can flush a finished buffer. + /// Route a slot message by status: confirmed advances the tip counter + /// and flushes the decoder, finalized advances the finalized watermark, + /// and any other status is dropped since its transactions may still be + /// in flight. async fn handle_slot( tx: &Sender, latest_chain_slot: &AtomicU64, slot: SubscribeUpdateSlot, ) -> ControlFlow<()> { - latest_chain_slot.fetch_max(slot.slot, Ordering::Relaxed); - Self::forward( - tx, - StreamUpdate::Slot { - slot: Slot(slot.slot), - }, - ) - .await + match slot.status() { + SlotStatus::SlotConfirmed => { + latest_chain_slot.fetch_max(slot.slot, Ordering::Relaxed); + Self::forward( + tx, + StreamUpdate::Slot { + slot: Slot(slot.slot), + }, + ) + .await + } + SlotStatus::SlotFinalized => { + Self::forward( + tx, + StreamUpdate::Finalized { + slot: Slot(slot.slot), + }, + ) + .await + } + _ => ControlFlow::Continue(()), + } } /// Push one update into the decoder channel. A full channel is the intended @@ -303,7 +320,7 @@ impl Ingester { /// matching updates, nothing routes on them today. const SETTLEMENT_FILTER: &str = "settlement_txs"; const SOLFLOW_FILTER: &str = "sol_flow_txs"; -const CHAIN_TIP_FILTER: &str = "chain_tip"; +const SLOT_FILTER: &str = "slot_statuses"; /// Where a fresh subscription starts. #[derive(Clone, Copy, Debug, PartialEq, Eq)] @@ -318,7 +335,7 @@ pub(crate) enum Resume { } /// The wire-level filter shape: the two named transaction filters and the -/// `chain_tip` slot filter, multiplexed into a single subscription at +/// slot-status filter, multiplexed into a single subscription at /// `confirmed` commitment. `from_slot` is the resume slot passed in by /// [`Ingester::serve`] (`last_indexed_slot + 1`, or `None` for the live tip). /// @@ -348,10 +365,12 @@ fn subscribe_request( SubscribeRequest { transactions: filters, slots: [( - CHAIN_TIP_FILTER.to_owned(), + SLOT_FILTER.to_owned(), SubscribeRequestFilterSlots { - // one message per slot at the subscription's commitment level - filter_by_commitment: Some(true), + // Every status transition, so finalized slots arrive next to + // confirmed ones. The ingester routes the two it needs and + // drops the rest. + filter_by_commitment: Some(false), ..Default::default() }, )] diff --git a/crates/solana-indexer/src/indexer/ingester/tests.rs b/crates/solana-indexer/src/indexer/ingester/tests.rs index 7fcef825ea..0695d1e046 100644 --- a/crates/solana-indexer/src/indexer/ingester/tests.rs +++ b/crates/solana-indexer/src/indexer/ingester/tests.rs @@ -5,6 +5,7 @@ use { channel::StreamUpdate, slot::Slot, wire::{ + SlotStatus, SubscribeUpdate, SubscribeUpdateAccount, SubscribeUpdateAccountInfo, @@ -70,10 +71,11 @@ fn account_update(slot: u64, sig: u8) -> Result { }) } -fn slot_update(slot: u64) -> Result { +fn slot_update(slot: u64, status: SlotStatus) -> Result { Ok(SubscribeUpdate { update_oneof: Some(UpdateOneof::Slot(SubscribeUpdateSlot { slot, + status: status as i32, ..Default::default() })), ..Default::default() @@ -127,8 +129,11 @@ async fn account_update_is_ignored() { } #[tokio::test] -async fn slot_update_advances_latest_chain_slot_and_is_forwarded() { - let (mut ingester, mut rx, slot) = ingester(stream::iter([slot_update(9_001)])); +async fn confirmed_slot_advances_latest_chain_slot_and_is_forwarded() { + let (mut ingester, mut rx, slot) = ingester(stream::iter([slot_update( + 9_001, + SlotStatus::SlotConfirmed, + )])); assert!(matches!(ingester.run().await, Err(Error::StreamEnded))); assert_eq!(slot.load(Ordering::Relaxed), 9_001); @@ -138,6 +143,38 @@ async fn slot_update_advances_latest_chain_slot_and_is_forwarded() { )); } +/// A finalized slot becomes the finalized-watermark signal. It does not +/// advance the chain-tip counter: the tip is the confirmed frontier. +#[tokio::test] +async fn finalized_slot_is_forwarded_without_moving_the_tip() { + let (mut ingester, mut rx, slot) = ingester(stream::iter([slot_update( + 8_970, + SlotStatus::SlotFinalized, + )])); + + assert!(matches!(ingester.run().await, Err(Error::StreamEnded))); + assert_eq!(slot.load(Ordering::Relaxed), 0); + assert!(matches!( + rx.try_recv(), + Ok(StreamUpdate::Finalized { slot: Slot(8_970) }) + )); + assert!(rx.is_empty()); +} + +/// Statuses ahead of the stream's commitment must not drive flushes: a +/// processed slot is dropped. +#[tokio::test] +async fn processed_slot_is_dropped() { + let (mut ingester, rx, slot) = ingester(stream::iter([slot_update( + 9_002, + SlotStatus::SlotProcessed, + )])); + + assert!(matches!(ingester.run().await, Err(Error::StreamEnded))); + assert_eq!(slot.load(Ordering::Relaxed), 0); + assert!(rx.is_empty()); +} + #[tokio::test] async fn unrelated_and_empty_updates_are_ignored() { let (mut ingester, mut rx, slot) = ingester(stream::iter([ diff --git a/crates/solana-indexer/src/indexer/mod.rs b/crates/solana-indexer/src/indexer/mod.rs index 62cf98ccfb..5e3b503702 100644 --- a/crates/solana-indexer/src/indexer/mod.rs +++ b/crates/solana-indexer/src/indexer/mod.rs @@ -1,6 +1,6 @@ //! Consumer components of the Solana settlement indexer. //! -//! The four components and their roles: +//! The two components and their roles: //! //! - [`Ingester`]: subscribes to the Yellowstone gRPC stream and drains it as //! fast as updates arrive, forwarding them to the decoder. It does no @@ -12,12 +12,9 @@ //! belonging to the settlement and SolFlow programs, and persists the //! resulting typed events to the store. //! -//! - [`FinalizationWorker`]: rows are first written at the `confirmed` -//! commitment level. This worker re-checks them against the chain and -//! promotes them to `finalized`, or marks them rolled back if the transaction -//! disappeared. It uses a cheap batched RPC call for recent rows and falls -//! back to one-call-per-row lookups for rows old enough that the batched -//! method no longer reports them. +//! Rows are written at the `confirmed` commitment level. The stream's +//! finalized slot statuses advance `solana.indexer_state.finalized_slot`, and +//! a row counts as final once its slot is at or below that watermark. pub mod decoder; pub mod ingester; diff --git a/crates/solana-indexer/src/persistence.rs b/crates/solana-indexer/src/persistence.rs index a931ce16e7..335d2ff170 100644 --- a/crates/solana-indexer/src/persistence.rs +++ b/crates/solana-indexer/src/persistence.rs @@ -342,6 +342,19 @@ WHERE indexer_state.slot < EXCLUDED.slot Ok(tx.commit().await?) } + /// Advance the finalized watermark. Update-only: before the first flush + /// there is no state row and nothing indexed to finalize. A backward + /// write is a no-op. + pub(crate) async fn write_finalized_slot(&self, slot: Slot) -> Result<(), PersistenceError> { + sqlx::query( + "UPDATE solana.indexer_state SET finalized_slot = GREATEST(finalized_slot, $1)", + ) + .bind(to_db_slot(slot)) + .execute(&self.pool) + .await?; + Ok(()) + } + /// Record a slot as fully indexed. A backward write is a no-op. pub(crate) async fn write_last_indexed_slot(&self, slot: Slot) -> Result<(), PersistenceError> { Self::upsert_last_indexed_slot(&self.pool, slot).await @@ -400,6 +413,34 @@ mod tests { std::collections::HashMap, }; + /// The finalized watermark only moves forward and needs an existing + /// state row: before the first flush the update is a no-op. + #[tokio::test] + #[ignore = "needs the solana.* schema applied locally, run with --test-threads 1"] + async fn solana_db_finalized_watermark_is_monotone_and_update_only() { + let pool = pool().await; + wipe(&pool).await; + let postgres = Postgres::new(pool.clone()); + + // No state row yet: the write lands nowhere. + postgres.write_finalized_slot(Slot(5)).await.unwrap(); + let row: Option = + sqlx::query_scalar("SELECT finalized_slot FROM solana.indexer_state") + .fetch_optional(&pool) + .await + .unwrap(); + assert_eq!(row, None); + + postgres.write_last_indexed_slot(Slot(10)).await.unwrap(); + postgres.write_finalized_slot(Slot(8)).await.unwrap(); + postgres.write_finalized_slot(Slot(6)).await.unwrap(); + let finalized: i64 = sqlx::query_scalar("SELECT finalized_slot FROM solana.indexer_state") + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(finalized, 8); + } + /// The `solana.orders` columns a seeded test order writes. struct SeedOrder { uid: [u8; 32], diff --git a/crates/solana-indexer/src/types/channel.rs b/crates/solana-indexer/src/types/channel.rs index 6c7c88585b..da3cfda4e3 100644 --- a/crates/solana-indexer/src/types/channel.rs +++ b/crates/solana-indexer/src/types/channel.rs @@ -19,13 +19,19 @@ pub(crate) enum StreamUpdate { /// Wire message body. inner: Box, }, - /// A slot-status message. Lets the decoder flush a buffered slot without - /// waiting for the next tracked transaction, which can be arbitrarily far - /// away. Only slots at the transaction stream's commitment may be - /// forwarded, an earlier-commitment slot would flush a buffer whose - /// transactions are still in flight. + /// A confirmed slot-status message. Lets the decoder flush a buffered + /// slot without waiting for the next tracked transaction, which can be + /// arbitrarily far away. Only slots at the transaction stream's + /// commitment may be forwarded, an earlier-commitment slot would flush a + /// buffer whose transactions are still in flight. Slot { /// The slot the status message reports. slot: Slot, }, + /// A finalized slot-status message. Advances the finalized watermark: + /// rows at or below it can no longer roll back. + Finalized { + /// The slot the status message reports finalized. + slot: Slot, + }, } diff --git a/database/sql-solana/V2__drop_chain_tip.sql b/database/sql-solana/V2__drop_chain_tip.sql new file mode 100644 index 0000000000..56e65f52cc --- /dev/null +++ b/database/sql-solana/V2__drop_chain_tip.sql @@ -0,0 +1,4 @@ +-- The chain tip duplicated the last-indexed watermark: quiet slots advance +-- solana.indexer_state.slot every confirmed slot, so freshness checks read +-- that row and this table had no reader left. +DROP TABLE solana.chain_tip;