solana-indexer: advance the finalized watermark from slot statuses

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(())
     }
 

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

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<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::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<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,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([

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;

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<i64> =
+            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],

crates/solana-indexer/src/types/channel.rs

@@ -19,13 +19,19 @@ pub(crate) enum StreamUpdate {
         /// Wire message body.
         inner: Box<SubscribeUpdateTransactionInfo>,
     },
-    /// 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,
+    },
 }

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;