solana-indexer: advance the finalized watermark from slot statuses

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

@@ -85,42 +85,47 @@ impl Decoder {
     /// instructions, and persists the results. Returns when the ingester
     /// drops the sender.
     ///
-    /// Events and dead letters are buffered per slot and flushed once a
-    /// transaction of a later slot arrives: stream resume is slot-granular
-    /// (`from_slot = last_indexed_slot + 1`), so the last indexed slot may
-    /// only name slots whose transactions have all been delivered. Buffering
-    /// also makes it one persistence batch per slot instead of one per
-    /// transaction.
+    /// Events and dead letters are buffered per slot and flushed when the
+    /// slot's confirmed status arrives: the stream delivers a slot's
+    /// transactions before that status, so the status is the completeness
+    /// signal. The watermark (`solana.indexer_state.slot`) advances to every
+    /// confirmed slot, quiet ones included, and only after its buffers
+    /// flushed. A persistence error aborts the decoder and the process: the
+    /// watermark did not advance past anything unflushed, so the restart
+    /// replays it.
     pub async fn run(&mut self) -> Result<(), PersistenceError> {
         let mut pending: BTreeMap<Slot, SlotBuffer> = BTreeMap::new();
-        let mut flushed_through: Option<Slot> = None;
+        // In-memory mirror of the persisted watermark, spares redundant
+        // writes and flags late transactions.
+        let mut watermark: Option<Slot> = None;
         while let Some(update) = self.rx.recv().await {
             let (slot, signature, inner) = match update {
                 StreamUpdate::Tx {
                     slot,
                     signature,
                     inner,
                 } => (slot, signature, inner),
-                // Slot statuses bound the flush latency: the next settlement
-                // transaction can be minutes away.
-                StreamUpdate::Slot { slot } => {
-                    self.flush_up_to(&mut pending, slot, &mut flushed_through)
+                StreamUpdate::Confirmed { slot } => {
+                    self.flush_confirmed(&mut pending, slot, &mut watermark)
                         .await?;
                     continue;
                 }
+                StreamUpdate::Finalized { slot } => {
+                    self.persistence.write_finalized_slot(slot).await?;
+                    continue;
+                }
             };
-            self.flush_up_to(&mut pending, slot, &mut flushed_through)
-                .await?;
 
-            if let Some(flushed) = flushed_through
-                && slot <= flushed
+            if let Some(watermark) = watermark
+                && slot <= watermark
             {
-                // The events below still persist, and the backward slot write
-                // write is a no-op, but a crash before this batch flushes
-                // would lose the transaction: resume starts past its slot.
+                // The provider broke the transactions-before-status ordering.
+                // The events below still persist (idempotent writes), but a
+                // crash before they flush would lose them: resume starts past
+                // their slot.
                 tracing::warn!(
                     %slot,
-                    flushed_through = %flushed,
+                    %watermark,
                     "transaction arrived for an already flushed slot"
                 );
             }
@@ -137,32 +142,34 @@ impl Decoder {
                 Err(DecodeFailed) => buffer.dead_letters.push(signature),
             }
         }
-        // The stream ended. Only the newest buffer may be missing
-        // transactions, everything below it was already proven complete by
-        // later stream activity.
-        let mut leftover = pending.into_iter().peekable();
-        while let Some((slot, buffer)) = leftover.next() {
-            let complete = leftover.peek().is_some();
-            self.flush_slot(slot, buffer, complete).await?;
+        // The stream ended before these buffers' confirmed statuses arrived,
+        // so they may be missing transactions: flush without advancing the
+        // watermark, the reconnect replays their slots.
+        for (slot, buffer) in pending {
+            self.flush_slot(slot, buffer, false).await?;
         }
         Ok(())
     }
 
-    /// Flushes every buffer at least [`FLUSH_HOLDBACK_SLOTS`] behind the
-    /// observed slot. The hold-back gives late-delivered transactions a
-    /// window to join their slot's still unflushed buffer, keeping them
-    /// crash-safe: resume replays everything past the last indexed slot.
-    async fn flush_up_to(
+    /// 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.
+    async fn flush_confirmed(
         &self,
         pending: &mut BTreeMap<Slot, SlotBuffer>,
-        observed: Slot,
-        flushed_through: &mut Option<Slot>,
+        confirmed: Slot,
+        watermark: &mut Option<Slot>,
     ) -> Result<(), PersistenceError> {
-        let cutoff = u64::from(observed).saturating_sub(FLUSH_HOLDBACK_SLOTS);
-        let keep = pending.split_off(&Slot(cutoff.saturating_add(1)));
-        for (slot, buffer) in std::mem::replace(pending, keep) {
+        while let Some(entry) = pending.first_entry() {
+            if *entry.key() > confirmed {
+                break;
+            }
+            let (slot, buffer) = entry.remove_entry();
             self.flush_slot(slot, buffer, true).await?;
-            *flushed_through = (*flushed_through).max(Some(slot));
+        }
+        if watermark.is_none_or(|watermark| watermark < confirmed) {
+            self.persistence.write_last_indexed_slot(confirmed).await?;
+            *watermark = Some(confirmed);
         }
         Ok(())
     }
@@ -307,11 +314,6 @@ impl Decoder {
     }
 }
 
-/// Slots stay buffered until the stream reports a slot this far past them.
-/// A transaction delivered up to this many slots late still joins its own
-/// unflushed buffer instead of racing the last-indexed-slot advance.
-const FLUSH_HOLDBACK_SLOTS: u64 = 2;
-
 /// One slot's accumulated output, flushed once the stream moves past the
 /// hold-back window.
 #[derive(Default)]

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()
@@ -712,22 +714,40 @@ 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.
+    // No confirmed status arrived yet, so nothing may flush: the buffers
+    // wait and the watermark stays unset.
     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);
-
-    // 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();
+    let events: i64 = sqlx::query_scalar("SELECT count(*) FROM solana.order_pda")
+        .fetch_one(&pool)
+        .await
+        .unwrap();
+    assert_eq!(events, 0);
+
+    // The confirmed 45 status flushes both buffered slots and advances the
+    // watermark to 45. The finalized status advances the finalized
+    // watermark, and the quiet confirmed 50 moves the last indexed slot to
+    // 50 with nothing to flush. 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(50)));
+    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

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 processed is dropped since those slots can still be skipped or
+    /// orphaned.
     async fn handle_slot(
         tx: &Sender<StreamUpdate>,
         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::Confirmed {
+                        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<GeyserStream> {
 /// 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()
             },
         )]

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<SubscribeUpdate, Status> {
     })
 }
 
-fn slot_update(slot: u64) -> Result<SubscribeUpdate, Status> {
+fn slot_update(slot: u64, status: SlotStatus) -> Result<SubscribeUpdate, Status> {
     Ok(SubscribeUpdate {
         update_oneof: Some(UpdateOneof::Slot(SubscribeUpdateSlot {
             slot,
+            status: status as i32,
             ..Default::default()
         })),
         ..Default::default()
@@ -127,17 +129,52 @@ 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);
     assert!(matches!(
         rx.try_recv(),
-        Ok(StreamUpdate::Slot { slot: Slot(9_001) })
+        Ok(StreamUpdate::Confirmed { slot: Slot(9_001) })
     ));
 }
 
+/// 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([