From 56bbde4d10fde48019e8f4f9fd466470ec721902 Mon Sep 17 00:00:00 2001 From: squadgazzz <22964585+squadgazzz@users.noreply.github.com> Date: Thu, 3 Sep 2026 11:44:33 +0000 Subject: [PATCH 1/4] Advance the finalized watermark and chain tip from slot statuses --- crates/solana-indexer/src/indexer/decoder.rs | 13 ++++ .../src/indexer/decoder/tests.rs | 38 ++++++++-- crates/solana-indexer/src/indexer/ingester.rs | 45 ++++++++--- .../src/indexer/ingester/tests.rs | 42 ++++++++++- crates/solana-indexer/src/indexer/mod.rs | 9 +-- crates/solana-indexer/src/persistence.rs | 75 +++++++++++++++++++ crates/solana-indexer/src/run.rs | 33 +++++++- crates/solana-indexer/src/types/channel.rs | 16 ++-- 8 files changed, 236 insertions(+), 35 deletions(-) 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..82b419c3d8 100644 --- a/crates/solana-indexer/src/indexer/ingester.rs +++ b/crates/solana-indexer/src/indexer/ingester.rs @@ -30,6 +30,7 @@ use { slot::Slot, wire::{ CommitmentLevel, + SlotStatus, SubscribeRequest, SubscribeRequestFilterSlots, SubscribeRequestFilterTransactions, @@ -197,21 +198,39 @@ 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. + /// Consume a slot message, routed by its status. A confirmed slot + /// advances the in-memory chain-tip counter and lets the decoder flush a + /// finished buffer. A finalized slot advances the finalized watermark. + /// Every other status is dropped: flushing on a slot ahead of the + /// transaction stream's commitment would declare slots complete whose + /// transactions are still 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 @@ -350,8 +369,10 @@ fn subscribe_request( slots: [( CHAIN_TIP_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..901fb89319 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,37 @@ 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) }) + )); +} + +/// 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..77129da4d2 100644 --- a/crates/solana-indexer/src/indexer/mod.rs +++ b/crates/solana-indexer/src/indexer/mod.rs @@ -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..cd771e34b4 100644 --- a/crates/solana-indexer/src/persistence.rs +++ b/crates/solana-indexer/src/persistence.rs @@ -342,6 +342,35 @@ 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 the last observed chain tip. Not monotone: a provider + /// reconnect can legitimately report an older tip. + pub(crate) async fn upsert_chain_tip(&self, slot: Slot) -> Result<(), PersistenceError> { + sqlx::query( + r#" +INSERT INTO solana.chain_tip (slot) +VALUES ($1) +ON CONFLICT (singleton) DO UPDATE SET slot = EXCLUDED.slot + "#, + ) + .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 +429,52 @@ 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 chain tip is a plain upsert: a reconnected provider may report an + /// older tip and the row follows it. + #[tokio::test] + #[ignore = "needs the solana.* schema applied locally, run with --test-threads 1"] + async fn solana_db_chain_tip_follows_the_last_write() { + let pool = pool().await; + wipe(&pool).await; + let postgres = Postgres::new(pool.clone()); + + postgres.upsert_chain_tip(Slot(100)).await.unwrap(); + postgres.upsert_chain_tip(Slot(90)).await.unwrap(); + let tip: i64 = sqlx::query_scalar("SELECT slot FROM solana.chain_tip") + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(tip, 90); + } + /// The `solana.orders` columns a seeded test order writes. struct SeedOrder { uid: [u8; 32], diff --git a/crates/solana-indexer/src/run.rs b/crates/solana-indexer/src/run.rs index b1843e3824..71e50e5dda 100644 --- a/crates/solana-indexer/src/run.rs +++ b/crates/solana-indexer/src/run.rs @@ -9,6 +9,7 @@ use { ingester::{Error, INGEST_TO_DECODER_CAPACITY, Ingester, Resume}, }, persistence::Postgres, + types::slot::Slot, yellowstone, }, clap::Parser, @@ -18,7 +19,10 @@ use { std::{ net::SocketAddr, path::PathBuf, - sync::{Arc, atomic::AtomicU64}, + sync::{ + Arc, + atomic::{AtomicU64, Ordering}, + }, time::Duration, }, tokio::{sync::mpsc, task::JoinHandle}, @@ -28,6 +32,9 @@ use { /// Wait between attempts to bring the stream back up. const STREAM_RETRY: Duration = Duration::from_secs(5); +/// How often the observed chain tip is written to the database. +const CHAIN_TIP_INTERVAL: Duration = Duration::from_secs(1); + /// The Solana indexer command line arguments. #[derive(Debug, Parser)] #[command(author, version, about)] @@ -84,6 +91,30 @@ async fn run(config: Config, start_slot: Option) { let mut decoder_task = tokio::spawn(async move { decoder.run().await }); let latest_chain_slot = Arc::new(AtomicU64::default()); + + // Samples the ingester's chain-tip counter: the tip stays fresh under + // decoder backpressure and the stream's hot path never writes to the + // database. + let chain_tip_loop = { + let persistence = persistence.clone(); + let latest_chain_slot = latest_chain_slot.clone(); + async move { + let mut written = 0; + loop { + tokio::time::sleep(CHAIN_TIP_INTERVAL).await; + let tip = latest_chain_slot.load(Ordering::Relaxed); + if tip <= written { + continue; + } + match persistence.upsert_chain_tip(Slot(tip)).await { + Ok(()) => written = tip, + Err(err) => tracing::warn!(?err, "chain tip write failed"), + } + } + } + }; + tokio::spawn(chain_tip_loop); + let stream_loop = async { let mut resume = start_slot.map_or(Resume::Watermark, Resume::From); loop { 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, + }, } From 61b1fa304fe64b8d102568e7d8130953d0f24c9e Mon Sep 17 00:00:00 2001 From: squadgazzz <22964585+squadgazzz@users.noreply.github.com> Date: Thu, 3 Sep 2026 11:57:06 +0000 Subject: [PATCH 2/4] Fix stale component docs --- crates/solana-indexer/src/indexer/ingester.rs | 2 +- crates/solana-indexer/src/indexer/ingester/tests.rs | 1 + crates/solana-indexer/src/indexer/mod.rs | 2 +- 3 files changed, 3 insertions(+), 2 deletions(-) diff --git a/crates/solana-indexer/src/indexer/ingester.rs b/crates/solana-indexer/src/indexer/ingester.rs index 82b419c3d8..7d28d29719 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 diff --git a/crates/solana-indexer/src/indexer/ingester/tests.rs b/crates/solana-indexer/src/indexer/ingester/tests.rs index 901fb89319..0695d1e046 100644 --- a/crates/solana-indexer/src/indexer/ingester/tests.rs +++ b/crates/solana-indexer/src/indexer/ingester/tests.rs @@ -158,6 +158,7 @@ async fn finalized_slot_is_forwarded_without_moving_the_tip() { 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 diff --git a/crates/solana-indexer/src/indexer/mod.rs b/crates/solana-indexer/src/indexer/mod.rs index 77129da4d2..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 From 1bbfdabc4ebd469bce895f218047238af0db0bd9 Mon Sep 17 00:00:00 2001 From: squadgazzz <22964585+squadgazzz@users.noreply.github.com> Date: Thu, 3 Sep 2026 17:56:55 +0000 Subject: [PATCH 3/4] Drop the chain tip table, the watermark is the freshness signal --- crates/solana-indexer/src/indexer/ingester.rs | 6 ++-- crates/solana-indexer/src/persistence.rs | 34 ------------------- crates/solana-indexer/src/run.rs | 33 +----------------- database/sql-solana/V2__drop_chain_tip.sql | 4 +++ 4 files changed, 8 insertions(+), 69 deletions(-) create mode 100644 database/sql-solana/V2__drop_chain_tip.sql diff --git a/crates/solana-indexer/src/indexer/ingester.rs b/crates/solana-indexer/src/indexer/ingester.rs index 7d28d29719..d2f892cb0d 100644 --- a/crates/solana-indexer/src/indexer/ingester.rs +++ b/crates/solana-indexer/src/indexer/ingester.rs @@ -322,7 +322,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)] @@ -337,7 +337,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). /// @@ -367,7 +367,7 @@ fn subscribe_request( SubscribeRequest { transactions: filters, slots: [( - CHAIN_TIP_FILTER.to_owned(), + SLOT_FILTER.to_owned(), SubscribeRequestFilterSlots { // Every status transition, so finalized slots arrive next to // confirmed ones. The ingester routes the two it needs and diff --git a/crates/solana-indexer/src/persistence.rs b/crates/solana-indexer/src/persistence.rs index cd771e34b4..335d2ff170 100644 --- a/crates/solana-indexer/src/persistence.rs +++ b/crates/solana-indexer/src/persistence.rs @@ -355,22 +355,6 @@ WHERE indexer_state.slot < EXCLUDED.slot Ok(()) } - /// Record the last observed chain tip. Not monotone: a provider - /// reconnect can legitimately report an older tip. - pub(crate) async fn upsert_chain_tip(&self, slot: Slot) -> Result<(), PersistenceError> { - sqlx::query( - r#" -INSERT INTO solana.chain_tip (slot) -VALUES ($1) -ON CONFLICT (singleton) DO UPDATE SET slot = EXCLUDED.slot - "#, - ) - .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 @@ -457,24 +441,6 @@ mod tests { assert_eq!(finalized, 8); } - /// The chain tip is a plain upsert: a reconnected provider may report an - /// older tip and the row follows it. - #[tokio::test] - #[ignore = "needs the solana.* schema applied locally, run with --test-threads 1"] - async fn solana_db_chain_tip_follows_the_last_write() { - let pool = pool().await; - wipe(&pool).await; - let postgres = Postgres::new(pool.clone()); - - postgres.upsert_chain_tip(Slot(100)).await.unwrap(); - postgres.upsert_chain_tip(Slot(90)).await.unwrap(); - let tip: i64 = sqlx::query_scalar("SELECT slot FROM solana.chain_tip") - .fetch_one(&pool) - .await - .unwrap(); - assert_eq!(tip, 90); - } - /// The `solana.orders` columns a seeded test order writes. struct SeedOrder { uid: [u8; 32], diff --git a/crates/solana-indexer/src/run.rs b/crates/solana-indexer/src/run.rs index 71e50e5dda..b1843e3824 100644 --- a/crates/solana-indexer/src/run.rs +++ b/crates/solana-indexer/src/run.rs @@ -9,7 +9,6 @@ use { ingester::{Error, INGEST_TO_DECODER_CAPACITY, Ingester, Resume}, }, persistence::Postgres, - types::slot::Slot, yellowstone, }, clap::Parser, @@ -19,10 +18,7 @@ use { std::{ net::SocketAddr, path::PathBuf, - sync::{ - Arc, - atomic::{AtomicU64, Ordering}, - }, + sync::{Arc, atomic::AtomicU64}, time::Duration, }, tokio::{sync::mpsc, task::JoinHandle}, @@ -32,9 +28,6 @@ use { /// Wait between attempts to bring the stream back up. const STREAM_RETRY: Duration = Duration::from_secs(5); -/// How often the observed chain tip is written to the database. -const CHAIN_TIP_INTERVAL: Duration = Duration::from_secs(1); - /// The Solana indexer command line arguments. #[derive(Debug, Parser)] #[command(author, version, about)] @@ -91,30 +84,6 @@ async fn run(config: Config, start_slot: Option) { let mut decoder_task = tokio::spawn(async move { decoder.run().await }); let latest_chain_slot = Arc::new(AtomicU64::default()); - - // Samples the ingester's chain-tip counter: the tip stays fresh under - // decoder backpressure and the stream's hot path never writes to the - // database. - let chain_tip_loop = { - let persistence = persistence.clone(); - let latest_chain_slot = latest_chain_slot.clone(); - async move { - let mut written = 0; - loop { - tokio::time::sleep(CHAIN_TIP_INTERVAL).await; - let tip = latest_chain_slot.load(Ordering::Relaxed); - if tip <= written { - continue; - } - match persistence.upsert_chain_tip(Slot(tip)).await { - Ok(()) => written = tip, - Err(err) => tracing::warn!(?err, "chain tip write failed"), - } - } - } - }; - tokio::spawn(chain_tip_loop); - let stream_loop = async { let mut resume = start_slot.map_or(Resume::Watermark, Resume::From); loop { 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; From be99aa81f5564fb698d1e18ab1f4722f2405ccb4 Mon Sep 17 00:00:00 2001 From: squadgazzz <22964585+squadgazzz@users.noreply.github.com> Date: Mon, 7 Sep 2026 09:06:31 +0000 Subject: [PATCH 4/4] Shrink the slot routing comment --- crates/solana-indexer/src/indexer/ingester.rs | 10 ++++------ 1 file changed, 4 insertions(+), 6 deletions(-) diff --git a/crates/solana-indexer/src/indexer/ingester.rs b/crates/solana-indexer/src/indexer/ingester.rs index d2f892cb0d..5c1641a5af 100644 --- a/crates/solana-indexer/src/indexer/ingester.rs +++ b/crates/solana-indexer/src/indexer/ingester.rs @@ -198,12 +198,10 @@ where .await } - /// Consume a slot message, routed by its status. A confirmed slot - /// advances the in-memory chain-tip counter and lets the decoder flush a - /// finished buffer. A finalized slot advances the finalized watermark. - /// Every other status is dropped: flushing on a slot ahead of the - /// transaction stream's commitment would declare slots complete whose - /// transactions are still in flight. + /// 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,