From 621e0d647500e769a3f8486d8de1ee5861bf231c Mon Sep 17 00:00:00 2001 From: senzenn Date: Tue, 1 Sep 2026 10:38:03 +0530 Subject: [PATCH 1/9] refactor(grpc-source): extract subscribe request construction The subscribe request was built inline in connect. Pull it into build_subscribe_request so there is one place that turns a Filters into a wire request, which is what a second caller will need. from_slot stays out of the builder. It is a one-time start position rather than a steady-state setting, which is why the client library overwrites the field with the live checkpoint on reconnect instead of reusing the configured value, so the initial subscribe sets it at the call site. --- crates/yellowstone-grpc-source/src/lib.rs | 86 ++++++++++++++++++++--- 1 file changed, 78 insertions(+), 8 deletions(-) diff --git a/crates/yellowstone-grpc-source/src/lib.rs b/crates/yellowstone-grpc-source/src/lib.rs index 3003fbe5..66b0e0eb 100644 --- a/crates/yellowstone-grpc-source/src/lib.rs +++ b/crates/yellowstone-grpc-source/src/lib.rs @@ -149,6 +149,24 @@ pub struct YellowstoneGrpcSource { config: YellowstoneGrpcConfig, } +/// Build the wire subscription for `filters`, layering on the commitment the +/// `From` conversion leaves unset. +/// +/// `from_slot` is deliberately not applied here. It is a one-time start +/// position rather than a steady-state setting, which is why the client +/// library overwrites the field with the live checkpoint on reconnect instead +/// of reusing the configured value. Only the initial subscribe sets it. +/// +fn build_subscribe_request(filters: Filters, config: &YellowstoneGrpcConfig) -> SubscribeRequest { + let mut request: SubscribeRequest = filters.into(); + + if let Some(commitment_level) = config.commitment_level { + request.commitment = Some(commitment_level as i32); + } + + request +} + #[async_trait] impl SourceTrait for YellowstoneGrpcSource { type Config = YellowstoneGrpcConfig; @@ -179,13 +197,8 @@ impl SourceTrait for YellowstoneGrpcSource { let mut client = builder.connect().await?; - let mut subscribe_request: SubscribeRequest = filters.into(); - if let Some(from_slot) = config.from_slot { - subscribe_request.from_slot = Some(from_slot); - } - if let Some(commitment_level) = config.commitment_level { - subscribe_request.commitment = Some(commitment_level as i32); - } + let mut subscribe_request = build_subscribe_request(filters, &config); + subscribe_request.from_slot = config.from_slot; tracing::debug!( has_transactions = !subscribe_request.transactions.is_empty(), @@ -238,7 +251,64 @@ impl SourceTrait for YellowstoneGrpcSource { #[cfg(test)] mod tests { - use super::YellowstoneGrpcConfig; + use std::collections::HashMap; + + use shipstern_core::Filters; + + use super::{build_subscribe_request, CommitmentLevel, YellowstoneGrpcConfig}; + + fn config_from(toml_src: &str) -> YellowstoneGrpcConfig { + toml::from_str(toml_src).expect("config must deserialize") + } + + /// The commitment `Filters` does not carry is layered on by the builder. + #[test] + fn subscribe_request_carries_commitment() { + let config = config_from( + r#" + endpoint = "https://example.rpcpool.com" + timeout = 60 + commitment-level = "finalized" + "#, + ); + + let request = build_subscribe_request(Filters::new(HashMap::new()), &config); + + assert_eq!(request.commitment, Some(CommitmentLevel::Finalized as i32)); + } + + /// A configured `from-slot` is a resume position, so the builder leaves it + /// alone and the initial subscribe sets it at the call site. + #[test] + fn subscribe_request_omits_from_slot() { + let config = config_from( + r#" + endpoint = "https://example.rpcpool.com" + timeout = 60 + from-slot = 350000000 + "#, + ); + + let request = build_subscribe_request(Filters::new(HashMap::new()), &config); + + assert_eq!(request.from_slot, None); + } + + /// Without a commitment the request keeps what the `Filters` conversion + /// produced, which leaves it unset. + #[test] + fn subscribe_request_omits_unset_commitment() { + let config = config_from( + r#" + endpoint = "https://example.rpcpool.com" + timeout = 60 + "#, + ); + + let request = build_subscribe_request(Filters::new(HashMap::new()), &config); + + assert_eq!(request.commitment, None); + } /// A config file predating the reconnect fields must still deserialize: /// missing `Option` keys become `None`, and the missing `auto-reconnect` From 73edba03711e9a36c5a863285b247be8fba7088c Mon Sep 17 00:00:00 2001 From: senzenn Date: Tue, 1 Sep 2026 10:39:34 +0530 Subject: [PATCH 2/9] feat(runtime): add a filter update channel to SourceTrait Sources hold their subscription for the lifetime of the connection, so a caller whose filter set changes has to drop the connection and reconnect to pick it up, losing messages in the window. Add the seam for changing it in place. SourceTrait gains two defaulted methods: supports_filter_updates, and connect_with_filter_updates which ignores the receiver and defers to connect. Defaults rather than a signature change on connect keeps all 17 existing implementations compiling untouched, 11 of which are test mocks. The builder creates the channel and Runtime hands out the sending half via filter_updates, gated on the source advertising support so a caller wiring updates to a source that ignores them finds out at the call site. The runtime drops its own copy of the sender before running, otherwise a source waiting on updates would wait forever on a channel nobody could send to. The payload is Filters rather than SubscribeRequest. The conversion to a wire request leaves commitment and from_slot unset and the gRPC source fills them from config afterwards, so a raw request would bypass that and let the subscription drift from the registered parsers. --- crates/runtime/src/builder.rs | 12 ++++++++++ crates/runtime/src/lib.rs | 41 ++++++++++++++++++++++++++++++++++- crates/runtime/src/sources.rs | 34 ++++++++++++++++++++++++++++- 3 files changed, 85 insertions(+), 2 deletions(-) diff --git a/crates/runtime/src/builder.rs b/crates/runtime/src/builder.rs index a2419798..43bb429f 100644 --- a/crates/runtime/src/builder.rs +++ b/crates/runtime/src/builder.rs @@ -3,6 +3,7 @@ use shipstern_core::{ instruction::InstructionUpdate, AccountUpdate, BlockMetaUpdate, BlockUpdate, SlotUpdate, TransactionUpdate, }; +use tokio::sync::mpsc; use crate::{ config::ShipsternConfig, @@ -12,6 +13,13 @@ use crate::{ util, Runtime, }; +/// Depth of the filter update channel handed to callers by +/// [`Runtime::filter_updates`]. Updates are rare, so a shallow queue is enough +/// to keep a caller from blocking on a short burst. The queue does not +/// coalesce: every set sent is forwarded, and the server applies them in order +/// and ends on the newest. +const FILTER_UPDATE_CHANNEL_SIZE: usize = 8; + /// Helper trait for defining the intended use for a builder. pub trait BuilderKind: Default { /// The type of error returned by the builder. @@ -265,10 +273,14 @@ impl RuntimeBuilder { return Err(BuilderError::SlotPipelineCollision); } + let (filter_updates_tx, filter_updates_rx) = mpsc::channel(FILTER_UPDATE_CHANNEL_SIZE); + Ok(Runtime { buffer: buffer_cfg, source: source_cfg, pipelines, + filter_updates_tx: Some(filter_updates_tx), + filter_updates_rx, _source: std::marker::PhantomData, #[cfg(feature = "prometheus")] metrics_registry, diff --git a/crates/runtime/src/lib.rs b/crates/runtime/src/lib.rs index 2bbeda41..b747d19b 100644 --- a/crates/runtime/src/lib.rs +++ b/crates/runtime/src/lib.rs @@ -15,6 +15,7 @@ use std::marker::PhantomData; use config::BufferConfig; +use shipstern_core::Filters; use tokio::sync::{mpsc, oneshot}; use yellowstone_grpc_proto::tonic::Status; @@ -83,6 +84,8 @@ pub struct Runtime { buffer: BufferConfig, source: S::Config, pipelines: handler::PipelineSets, + filter_updates_tx: Option>, + filter_updates_rx: mpsc::Receiver, #[cfg(feature = "prometheus")] metrics_registry: prometheus::Registry, _source: PhantomData, @@ -91,6 +94,34 @@ pub struct Runtime { impl Runtime { /// Create a new runtime builder. pub fn builder() -> RuntimeBuilder { RuntimeBuilder::::default() } + + /// Take the sending half of the filter update channel. + /// + /// Each [`Filters`] sent replaces the whole subscription rather than + /// adding to it, because that is what the gRPC servers do with a + /// mid-stream request, so send the complete set every time. + /// + /// Keys are parser IDs. A key matching no registered pipeline still + /// changes the wire subscription, so the server streams that data and the + /// runtime then drops every update of it with only a trace-level line, so + /// take the keys from the parsers actually registered on this runtime. + /// + /// Delivery is best effort. A set rejected while the source is between + /// connections is retried once the stream recovers, but nothing reports + /// back to the sender either way. + /// + /// Returns `None` when the source does not support filter updates, or + /// when the sender has already been taken. Call this before running the + /// runtime, since [`Self::run`], [`Self::try_run`], [`Self::run_async`] + /// and [`Self::try_run_async`] all consume it. + /// + pub fn filter_updates(&mut self) -> Option> { + if !S::supports_filter_updates() { + return None; + } + + self.filter_updates_tx.take() + } } impl Runtime { /// Create a new Tokio runtime and run the Shipstern runtime within it, @@ -252,9 +283,17 @@ impl Runtime { let filters = self.pipelines.filters(); let source = S::new(self.source, filters); + let filter_updates = self.filter_updates_rx; + + // Close the channel when nobody asked for the sending half, so a source + // that waits on updates is not left waiting on a sender that can never + // produce one. + drop(self.filter_updates_tx); tokio::spawn(async move { - let _ = source.connect(tx, status_tx).await; + let _ = source + .connect_with_filter_updates(tx, status_tx, filter_updates) + .await; }); let signal; diff --git a/crates/runtime/src/sources.rs b/crates/runtime/src/sources.rs index 42bd2c16..bde2e47a 100644 --- a/crates/runtime/src/sources.rs +++ b/crates/runtime/src/sources.rs @@ -5,7 +5,10 @@ use async_trait::async_trait; use shipstern_core::Filters; -use tokio::sync::{mpsc::Sender, oneshot}; +use tokio::sync::{ + mpsc::{Receiver, Sender}, + oneshot, +}; use yellowstone_grpc_proto::{geyser::SubscribeUpdate, tonic}; /// How a source exited. @@ -45,4 +48,33 @@ pub trait SourceTrait: std::fmt::Debug + Send + Sync + 'static { tx: Sender>, status_tx: oneshot::Sender, ) -> Result<(), crate::Error>; + + /// Whether this source applies filter updates to a live subscription. + /// + /// The runtime checks this before handing a caller the sending half of the + /// filter update channel, so a caller wiring updates to a source that + /// ignores them finds out at the call site instead of silently sending + /// into a void. + /// + #[must_use] + fn supports_filter_updates() -> bool { false } + + /// Connect and stream updates, applying filter sets received on + /// `filter_updates` to the live subscription. + /// + /// The default ignores `filter_updates` and defers to [`Self::connect`], + /// so a source that cannot change its subscription mid-stream needs no + /// implementation. Override this together with + /// [`Self::supports_filter_updates`]. + /// + async fn connect_with_filter_updates( + &self, + tx: Sender>, + status_tx: oneshot::Sender, + filter_updates: Receiver, + ) -> Result<(), crate::Error> { + drop(filter_updates); + + self.connect(tx, status_tx).await + } } From b41c019bf2ff12a7f3f5802d53eeca1fbcf3fb48 Mon Sep 17 00:00:00 2001 From: senzenn Date: Tue, 1 Sep 2026 10:40:14 +0530 Subject: [PATCH 3/9] feat(grpc-source): apply filter updates to the live subscription subscribe_with_request returns a sink alongside the stream, and the source bound it to _sub_tx and dropped it. Keep it and drive it from the update channel, so the read loop becomes a select over the stream and the receiver. Both trait methods funnel into one private run, so there is a single connection path rather than two copies of it. A rejected send is held and retried rather than logged and dropped. Auto-reconnect is on by default and the client swallows recoverable stream errors to reconnect behind our back, so a send failing mid-reconnect never surfaces on the stream. Worse, the sink records a request into its reconnect state only after a successful send, so the reconnect would come back with the previous filters while the caller's send had already returned Ok. The stream producing again is the signal that the reconnect landed and the sink has a live sender, so that is when a held set goes out. Also spell out in build_subscribe_request why from_slot must not ride along on an update: geyser reads it as a replay request and either replays every slot since or ends the stream when it has no replay buffer, and richat rejects the request outright when the set contains blocks. The update branch retires itself once the sending half is gone, since a closed receiver is ready on every poll and would otherwise spin the loop. --- crates/yellowstone-grpc-source/Cargo.toml | 2 +- crates/yellowstone-grpc-source/src/lib.rs | 163 ++++++++++++++++++---- 2 files changed, 137 insertions(+), 28 deletions(-) diff --git a/crates/yellowstone-grpc-source/Cargo.toml b/crates/yellowstone-grpc-source/Cargo.toml index b3638a1a..cce7ed42 100644 --- a/crates/yellowstone-grpc-source/Cargo.toml +++ b/crates/yellowstone-grpc-source/Cargo.toml @@ -10,7 +10,7 @@ readme = "./README.md" [dependencies] async-trait = { workspace = true } -tokio = { workspace = true, features = ["rt-multi-thread", "signal"] } +tokio = { workspace = true, features = ["rt-multi-thread", "signal", "macros"] } tracing = { workspace = true } futures-util = { workspace = true, features = ["sink"] } shipstern = { workspace = true } diff --git a/crates/yellowstone-grpc-source/src/lib.rs b/crates/yellowstone-grpc-source/src/lib.rs index 66b0e0eb..e4f1b7e5 100644 --- a/crates/yellowstone-grpc-source/src/lib.rs +++ b/crates/yellowstone-grpc-source/src/lib.rs @@ -2,14 +2,17 @@ use std::time::Duration; use async_trait::async_trait; use clap::ValueEnum; -use futures_util::StreamExt; +use futures_util::{SinkExt, StreamExt}; use shipstern::{ sources::{SourceExitStatus, SourceTrait}, CommitmentLevel, Error as ShipsternError, }; use shipstern_core::Filters; -use tokio::sync::{mpsc::Sender, oneshot}; -use yellowstone_grpc_client::{Backoff, GeyserGrpcClient, ReconnectConfig}; +use tokio::sync::{ + mpsc::{self, Receiver, Sender}, + oneshot, +}; +use yellowstone_grpc_client::{Backoff, GeyserGrpcClient, ReconnectConfig, SubscribeRequestSink}; use yellowstone_grpc_proto::{ geyser::{SubscribeRequest, SubscribeUpdate}, tonic::{codec::CompressionEncoding, transport::ClientTlsConfig, Status}, @@ -153,9 +156,14 @@ pub struct YellowstoneGrpcSource { /// `From` conversion leaves unset. /// /// `from_slot` is deliberately not applied here. It is a one-time start -/// position rather than a steady-state setting, which is why the client -/// library overwrites the field with the live checkpoint on reconnect instead -/// of reusing the configured value. Only the initial subscribe sets it. +/// position rather than a steady-state setting, and repeating it on a +/// mid-stream update is destructive: yellowstone-grpc-geyser treats it as a +/// replay request and either replays every slot since, or ends the stream +/// with `from_slot is not supported` when it has no replay buffer, while +/// richat rejects the request outright if the set contains blocks. The client +/// library agrees, overwriting the field with the live checkpoint on +/// reconnect rather than reusing the configured value. Only the initial +/// subscribe sets it. /// fn build_subscribe_request(filters: Filters, config: &YellowstoneGrpcConfig) -> SubscribeRequest { let mut request: SubscribeRequest = filters.into(); @@ -167,16 +175,87 @@ fn build_subscribe_request(filters: Filters, config: &YellowstoneGrpcConfig) -> request } +/// Send `filters` to the server, handing it back unsent if the sink rejects it. +/// +/// A rejection means the request channel is disconnected. With auto-reconnect +/// enabled, which is the default, that is a transient reconnect window rather +/// than a fatal error: the stream yields nothing during it and the client +/// library swaps a fresh sender into this sink once it recovers. Dropping the +/// set here would lose it silently, because the sink only records a request +/// into its reconnect state after a successful send, so the reconnect would +/// resubscribe with the previous filters. +/// +async fn send_filter_update( + sink: &mut SubscribeRequestSink, + config: &YellowstoneGrpcConfig, + filters: Filters, +) -> Option { + let request = build_subscribe_request(filters.clone(), config); + + tracing::debug!( + accounts = request.accounts.len(), + transactions = request.transactions.len(), + slots = request.slots.len(), + "Sending filter update to the live subscription" + ); + + if let Err(err) = sink.send(request).await { + tracing::warn!( + %err, + "Filter update rejected by the sink, holding it until the stream recovers" + ); + + return Some(filters); + } + + None +} + #[async_trait] impl SourceTrait for YellowstoneGrpcSource { type Config = YellowstoneGrpcConfig; fn new(config: Self::Config, filters: Filters) -> Self { Self { config, filters } } + fn supports_filter_updates() -> bool { true } + async fn connect( &self, tx: Sender>, status_tx: oneshot::Sender, + ) -> Result<(), ShipsternError> { + let closed = { + let (_, rx) = mpsc::channel(1); + rx + }; + + self.run(tx, status_tx, closed).await + } + + async fn connect_with_filter_updates( + &self, + tx: Sender>, + status_tx: oneshot::Sender, + filter_updates: Receiver, + ) -> Result<(), ShipsternError> { + self.run(tx, status_tx, filter_updates).await + } +} + +impl YellowstoneGrpcSource { + /// Open the subscription and pump updates until the stream ends, sending + /// any filter set that arrives on `filter_updates` to the server so it + /// replaces the live subscription. + /// + /// An already closed `filter_updates` is expected rather than an error: + /// it is what `connect` passes when the caller never asked for updates, + /// and it retires the update branch after one poll. + /// + async fn run( + &self, + tx: Sender>, + status_tx: oneshot::Sender, + mut filter_updates: Receiver, ) -> Result<(), ShipsternError> { let filters = self.filters.clone(); let config = self.config.clone(); @@ -212,7 +291,7 @@ impl SourceTrait for YellowstoneGrpcSource { "Subscribing to gRPC stream" ); - let (_sub_tx, stream) = client + let (mut sink, stream) = client .subscribe_with_request(Some(subscribe_request)) .await?; @@ -220,25 +299,51 @@ impl SourceTrait for YellowstoneGrpcSource { tracing::debug!("gRPC stream started"); + let mut accept_filter_updates = true; + let mut pending_filters: Option = None; + let exit_status = loop { - match stream.next().await { - Some(Ok(update)) => { - if tx.send(Ok(update)).await.is_err() { - tracing::info!("Receiver dropped, stopping source"); - // Defensive only - normally unreachable because Signal/Buffer - // branch wins first when receiver drops. - break SourceExitStatus::ReceiverDropped; - } - }, - Some(Err(status)) => { - tracing::warn!(code = ?status.code(), message = %status.message(), "Received error status from stream"); - let code = status.code(); - let message = status.message().to_string(); - let _ = tx.send(Err(status)).await; - break SourceExitStatus::StreamError { code, message }; + tokio::select! { + update = stream.next() => match update { + Some(Ok(update)) => { + if tx.send(Ok(update)).await.is_err() { + tracing::info!("Receiver dropped, stopping source"); + // Defensive only - normally unreachable because Signal/Buffer + // branch wins first when receiver drops. + break SourceExitStatus::ReceiverDropped; + } + + // The stream producing again is the signal that any + // reconnect has landed and the sink has a live sender, + // so this is the moment to retry a held filter set. + if let Some(filters) = pending_filters.take() { + pending_filters = + send_filter_update(&mut sink, &config, filters).await; + } + }, + Some(Err(status)) => { + tracing::warn!(code = ?status.code(), message = %status.message(), "Received error status from stream"); + let code = status.code(); + let message = status.message().to_string(); + let _ = tx.send(Err(status)).await; + break SourceExitStatus::StreamError { code, message }; + }, + None => { + break SourceExitStatus::StreamEnded; + }, }, - None => { - break SourceExitStatus::StreamEnded; + + update = filter_updates.recv(), if accept_filter_updates => { + let Some(filters) = update else { + // Nobody holds the sending half any more, so leave this + // branch alone for the rest of the connection. + accept_filter_updates = false; + continue; + }; + + // A newer set supersedes anything still held, because every + // set is complete rather than a delta. + pending_filters = send_filter_update(&mut sink, &config, filters).await; }, } }; @@ -261,7 +366,9 @@ mod tests { toml::from_str(toml_src).expect("config must deserialize") } - /// The commitment `Filters` does not carry is layered on by the builder. + /// The startup subscribe and every later filter update are built by the + /// same function, so the commitment `Filters` does not carry lands on both + /// rather than only on the initial request. #[test] fn subscribe_request_carries_commitment() { let config = config_from( @@ -277,8 +384,10 @@ mod tests { assert_eq!(request.commitment, Some(CommitmentLevel::Finalized as i32)); } - /// A configured `from-slot` is a resume position, so the builder leaves it - /// alone and the initial subscribe sets it at the call site. + /// A configured `from-slot` must never reach a mid-stream update. Servers + /// read it as a replay request, so repeating it would replay the whole gap + /// or end the stream outright. Only the initial subscribe sets it, and it + /// is set at that call site rather than here. #[test] fn subscribe_request_omits_from_slot() { let config = config_from( From b538c4b3eeb8093bba8b6544befef4d61ad55333 Mon Sep 17 00:00:00 2001 From: senzenn Date: Tue, 1 Sep 2026 10:40:38 +0530 Subject: [PATCH 4/9] test(runtime): cover filter update delivery A mock source that records the filter sets it receives, covering the four cases the plumbing can get wrong: a set sent through the handle reaches the source, an unsupporting source hands out no sender, the sender is handed out at most once, and a runtime whose handle was never taken still runs to completion. The last one is a regression guard. try_run_async consumes the runtime but the unclaimed sender used to stay alive inside it for the whole scope, so a source awaiting recv waited on a channel that could never produce. It hangs without the explicit drop. Tests share one static for the recorded IDs and libtest runs them on separate threads, so each records under its own parser ID and asserts on that alone rather than clearing and comparing the whole vector. --- crates/runtime/src/runtime_tests.rs | 126 +++++++++++++++++++++++++++- 1 file changed, 123 insertions(+), 3 deletions(-) diff --git a/crates/runtime/src/runtime_tests.rs b/crates/runtime/src/runtime_tests.rs index 381848ce..322a9bbc 100644 --- a/crates/runtime/src/runtime_tests.rs +++ b/crates/runtime/src/runtime_tests.rs @@ -1,12 +1,19 @@ use std::{ borrow::Cow, - sync::atomic::{AtomicUsize, Ordering}, + collections::HashMap, + sync::{ + atomic::{AtomicUsize, Ordering}, + Mutex, + }, time::Duration, }; use async_trait::async_trait; -use shipstern_core::{ParseResult, Parser, Prefilter, SlotUpdate}; -use tokio::sync::{mpsc::Sender, oneshot}; +use shipstern_core::{Filters, ParseResult, Parser, Prefilter, SlotUpdate}; +use tokio::sync::{ + mpsc::{Receiver, Sender}, + oneshot, +}; use yellowstone_grpc_proto::{ geyser::{subscribe_update::UpdateOneof, SlotStatus, SubscribeUpdate, SubscribeUpdateSlot}, tonic, @@ -373,6 +380,119 @@ impl Handler for SlowSlotHandler { } } +/// Parser IDs carried by the filter sets a source received through the +/// runtime filter update channel. Tests run in parallel and all share this, +/// so each one records under its own ID and asserts on that ID alone. +static RECEIVED_FILTER_IDS: Mutex> = Mutex::new(Vec::new()); + +/// Parser ID used only by [`test_filter_updates_reach_a_supporting_source`]. +const SWAPPED_PARSER_ID: &str = "test::SwappedParser"; + +#[derive(Debug)] +struct MockFilterUpdateSource; + +#[async_trait] +impl SourceTrait for MockFilterUpdateSource { + type Config = NullConfig; + + fn new(_: NullConfig, _: Filters) -> Self { Self } + + fn supports_filter_updates() -> bool { true } + + async fn connect( + &self, + tx: Sender>, + status_tx: oneshot::Sender, + ) -> Result<(), Error> { + wait_for_runtime_ready().await; + signal_stream_ended(status_tx); + hold_channel_open_briefly().await; + drop(tx); + Ok(()) + } + + async fn connect_with_filter_updates( + &self, + tx: Sender>, + status_tx: oneshot::Sender, + mut filter_updates: Receiver, + ) -> Result<(), Error> { + wait_for_runtime_ready().await; + + if let Some(filters) = filter_updates.recv().await { + let mut ids = filters.parsers_filters.keys().cloned().collect::>(); + ids.sort(); + + RECEIVED_FILTER_IDS.lock().unwrap().extend(ids); + } + + signal_stream_ended(status_tx); + hold_channel_open_briefly().await; + drop(tx); + Ok(()) + } +} + +fn filters_for(ids: &[&str]) -> Filters { + Filters::new( + ids.iter() + .map(|id| ((*id).to_owned(), Prefilter::default())) + .collect::>(), + ) +} + +#[tokio::test] +async fn test_filter_updates_reach_a_supporting_source() { + let mut runtime = Runtime::::builder() + .try_build(default_test_config()) + .unwrap(); + + let updates = runtime + .filter_updates() + .expect("source advertises filter update support"); + + updates + .send(filters_for(&[SWAPPED_PARSER_ID])) + .await + .unwrap(); + + assert_server_hangup(runtime.try_run_async().await); + + let received = RECEIVED_FILTER_IDS.lock().unwrap().clone(); + assert!( + received.contains(&SWAPPED_PARSER_ID.to_owned()), + "source never saw the updated filter set, recorded {received:?}" + ); +} + +#[tokio::test] +async fn test_filter_updates_unavailable_when_source_does_not_support_them() { + let mut runtime = Runtime::::builder() + .try_build(default_test_config()) + .unwrap(); + + assert!(runtime.filter_updates().is_none()); +} + +#[tokio::test] +async fn test_filter_updates_handed_out_only_once() { + let mut runtime = Runtime::::builder() + .try_build(default_test_config()) + .unwrap(); + + assert!(runtime.filter_updates().is_some()); + assert!(runtime.filter_updates().is_none()); +} + +#[tokio::test] +async fn test_source_runs_when_the_filter_update_handle_is_never_taken() { + let runtime = Runtime::::builder() + .try_build(default_test_config()) + .unwrap(); + + assert_server_hangup(runtime.try_run_async().await); +} + #[tokio::test] async fn test_stream_end_returns_error() { let runtime = Runtime::::builder() From fd970d90b9f54fa333aa3b9babfc92e922870a9d Mon Sep 17 00:00:00 2001 From: senzenn Date: Tue, 1 Sep 2026 10:40:38 +0530 Subject: [PATCH 5/9] docs: note provider support for runtime filter updates Document how to take the handle before the runtime is consumed, that each set replaces the whole subscription rather than adding to it, and that keys are parser IDs so an unregistered one changes what the server sends and then gets discarded at trace level. Say plainly that the feature depends on the provider honouring mid-stream requests. Both yellowstone-grpc-geyser and richat apply them in their source, but a deployment can sit behind infrastructure that does not forward them. --- crates/yellowstone-grpc-source/README.md | 39 ++++++++++++++++++++++++ 1 file changed, 39 insertions(+) diff --git a/crates/yellowstone-grpc-source/README.md b/crates/yellowstone-grpc-source/README.md index 8954968d..0cea248a 100644 --- a/crates/yellowstone-grpc-source/README.md +++ b/crates/yellowstone-grpc-source/README.md @@ -19,6 +19,45 @@ The [`Source`](https://github.com/solana-rpc/shipstern/blob/main/crates/runtime/ - Configure filters for data processing - Manage source-specific configuration +## Runtime filter updates + +The gRPC source keeps the `SubscribeRequest` sink that `yellowstone-grpc-client` +returns next to the update stream, so a caller can change the subscription +without tearing down the connection and losing messages during the reconnect. + +Take the sender off the runtime before running it, because `run` and +`run_async` both consume the runtime: + +```rust +let mut runtime = Runtime::::builder() + .instruction(Pipeline::new(TokenProgramIxParser, [Handler])) + .try_build(config)?; + +if let Some(filter_updates) = runtime.filter_updates() { + tokio::spawn(async move { + filter_updates.send(new_filters).await.ok(); + }); +} + +runtime.run_async().await; +``` + +Each `Filters` sent replaces the whole subscription rather than adding to it, +which is what the server does with a mid-stream request, so send the complete +set every time. + +The map keys are parser IDs. A key that matches no registered pipeline still +changes what the server sends, and the runtime then discards all of it at trace +level, so take the keys from the parsers you registered. Delivery is best +effort: a set rejected while the source is between connections is retried once +the stream recovers, but the sender is not told either way. + +`Runtime::filter_updates` returns `None` for sources that do not implement +this, which today is every source except gRPC. Whether an update takes effect +also depends on the provider. Both `yellowstone-grpc-geyser` and `richat` apply +mid-stream requests to a live subscription, but a deployment can sit behind +infrastructure that does not forward them. + ## Creating a Custom Source Here's a step-by-step guide to creating your own source: From 952742b2a0bd196f3a4d8529d9583523908800cf Mon Sep 17 00:00:00 2001 From: senzenn Date: Tue, 1 Sep 2026 10:45:35 +0530 Subject: [PATCH 6/9] fix(grpc-source): make a server-rejected filter update diagnosable A server that refuses a filter set answers on the stream, not the sink, and the codes it uses are outside the client's recoverable set, so the stream ends and the runtime exits. That was already true before this branch for the initial subscribe, but sending sets mid-stream makes it reachable long after startup, where the cause is far less obvious. Nothing can un-send the request, so record how many updates went out on the connection and report it alongside the status. An operator seeing a run stop on InvalidArgument can then tell a rejected update apart from an unrelated server error. Document the same on the accessor and in the README. --- crates/runtime/src/lib.rs | 6 ++++++ crates/yellowstone-grpc-source/README.md | 6 ++++++ crates/yellowstone-grpc-source/src/lib.rs | 21 ++++++++++++++++++++- 3 files changed, 32 insertions(+), 1 deletion(-) diff --git a/crates/runtime/src/lib.rs b/crates/runtime/src/lib.rs index b747d19b..7cef5cd1 100644 --- a/crates/runtime/src/lib.rs +++ b/crates/runtime/src/lib.rs @@ -110,6 +110,12 @@ impl Runtime { /// connections is retried once the stream recovers, but nothing reports /// back to the sender either way. /// + /// A set the server itself refuses, by exceeding its configured filter + /// limits for example, is answered on the stream with a code the client + /// does not treat as recoverable, which ends the run. Sending a set the + /// provider will not accept therefore stops the runtime rather than + /// leaving the previous subscription in place. + /// /// Returns `None` when the source does not support filter updates, or /// when the sender has already been taken. Call this before running the /// runtime, since [`Self::run`], [`Self::try_run`], [`Self::run_async`] diff --git a/crates/yellowstone-grpc-source/README.md b/crates/yellowstone-grpc-source/README.md index 0cea248a..333f680d 100644 --- a/crates/yellowstone-grpc-source/README.md +++ b/crates/yellowstone-grpc-source/README.md @@ -52,6 +52,12 @@ level, so take the keys from the parsers you registered. Delivery is best effort: a set rejected while the source is between connections is retried once the stream recovers, but the sender is not told either way. +A set the server refuses, by exceeding its configured filter limits for +example, comes back on the stream with a code the client does not retry, which +ends the run. An update the provider will not accept stops the runtime rather +than leaving the previous subscription in place, so validate against the +provider's limits before sending one. + `Runtime::filter_updates` returns `None` for sources that do not implement this, which today is every source except gRPC. Whether an update takes effect also depends on the provider. Both `yellowstone-grpc-geyser` and `richat` apply diff --git a/crates/yellowstone-grpc-source/src/lib.rs b/crates/yellowstone-grpc-source/src/lib.rs index e4f1b7e5..1ffb0185 100644 --- a/crates/yellowstone-grpc-source/src/lib.rs +++ b/crates/yellowstone-grpc-source/src/lib.rs @@ -301,6 +301,7 @@ impl YellowstoneGrpcSource { let mut accept_filter_updates = true; let mut pending_filters: Option = None; + let mut filter_updates_sent: u64 = 0; let exit_status = loop { tokio::select! { @@ -319,10 +320,24 @@ impl YellowstoneGrpcSource { if let Some(filters) = pending_filters.take() { pending_filters = send_filter_update(&mut sink, &config, filters).await; + + if pending_filters.is_none() { + filter_updates_sent += 1; + } } }, Some(Err(status)) => { - tracing::warn!(code = ?status.code(), message = %status.message(), "Received error status from stream"); + // A server that rejects a filter set answers on the stream + // rather than the sink, and the codes it uses for that are + // not recoverable, so this is where a bad update surfaces. + // Report the count so an operator can tell that apart from + // an unrelated server error. + tracing::warn!( + code = ?status.code(), + message = %status.message(), + filter_updates_sent, + "Received error status from stream" + ); let code = status.code(); let message = status.message().to_string(); let _ = tx.send(Err(status)).await; @@ -344,6 +359,10 @@ impl YellowstoneGrpcSource { // A newer set supersedes anything still held, because every // set is complete rather than a delta. pending_filters = send_filter_update(&mut sink, &config, filters).await; + + if pending_filters.is_none() { + filter_updates_sent += 1; + } }, } }; From fc748bcd0b68b0cc033d99563c92f8830771d114 Mon Sep 17 00:00:00 2001 From: senzenn Date: Tue, 1 Sep 2026 11:35:12 +0530 Subject: [PATCH 7/9] docs: describe when a filter update becomes visible The server applies a new set promptly, but a consumer sees it only once whatever it has already queued drains, so the wait is a property of how far behind the pipeline is rather than of the update itself. Updates matching the new set arrive stale by that margin and settle to real time once the backlog clears. Say that plainly on the accessor and in the README, because a caller who reads the delay the other way will size the wrong problem. --- crates/runtime/src/lib.rs | 13 +++++++++++++ crates/yellowstone-grpc-source/README.md | 9 +++++++++ 2 files changed, 22 insertions(+) diff --git a/crates/runtime/src/lib.rs b/crates/runtime/src/lib.rs index 7cef5cd1..9f0a4571 100644 --- a/crates/runtime/src/lib.rs +++ b/crates/runtime/src/lib.rs @@ -106,6 +106,19 @@ impl Runtime { /// runtime then drops every update of it with only a trace-level line, so /// take the keys from the parsers actually registered on this runtime. /// + /// The server applies the new set promptly, but a consumer sees it only + /// once whatever it has already queued drains, so the delay is however far + /// behind the pipeline already was rather than a property of the update. + /// Measured against a live endpoint, a consumer running about 15 seconds + /// behind kept receiving the old set for roughly that long after the send, + /// and the first updates matching the new set arrived stale by the same + /// margin before catching up to real time. A pipeline keeping pace sees + /// the change almost at once. + /// + /// Treat a returned `send` as the request having been handed off, not as + /// the subscription having changed, and keep handlers able to cope with + /// updates matching the old set until the backlog clears. + /// /// Delivery is best effort. A set rejected while the source is between /// connections is retried once the stream recovers, but nothing reports /// back to the sender either way. diff --git a/crates/yellowstone-grpc-source/README.md b/crates/yellowstone-grpc-source/README.md index 333f680d..2a55a747 100644 --- a/crates/yellowstone-grpc-source/README.md +++ b/crates/yellowstone-grpc-source/README.md @@ -52,6 +52,15 @@ level, so take the keys from the parsers you registered. Delivery is best effort: a set rejected while the source is between connections is retried once the stream recovers, but the sender is not told either way. +The server applies the new set promptly, but you see it only once whatever is +already queued drains, so the delay is however far behind your pipeline already +was rather than a property of the update. Against a live endpoint, a consumer +running about 15 seconds behind kept receiving the old set for roughly that +long, and the first updates matching the new set arrived stale by the same +margin before catching up. A pipeline keeping pace sees the change almost at +once. A returned `send` means the request was handed off, not that the +subscription has changed. + A set the server refuses, by exceeding its configured filter limits for example, comes back on the stream with a code the client does not retry, which ends the run. An update the provider will not accept stops the runtime rather From 159573f06aa4bcb768286ae70d879e058f4bac1d Mon Sep 17 00:00:00 2001 From: senzenn Date: Tue, 1 Sep 2026 12:04:08 +0530 Subject: [PATCH 8/9] fix(grpc-source): retry a held filter set on a timer, not on stream traffic Review caught that a set held after a sink rejection was only retried inside the arm handling a successful stream item, which assumes the connection keeps producing. It need not. The filter being replaced may match nothing, and the client swallows its own keepalive messages rather than yielding them, so there are live connections where the retry site is never reached and the set is stranded for the rest of the connection. Retry on a five second tick instead, gated on something actually being held so the branch is disabled in the normal case. Log when a connection ends with a set still unsent, since the earlier warning promised a retry that then never happened. Also correct three things review found stale or wrong. The channel-depth doc said sets are never coalesced, which stopped being true when a newer set started superseding a held one. The accessor doc said a refused set ends the run, understating it: run and run_async treat that error as fatal and exit the process. And the delivery test built its payload from Prefilter::default(), which converts to an entirely empty request, so it would have passed on a payload that says nothing on the wire. --- crates/runtime/src/builder.rs | 9 ++++--- crates/runtime/src/lib.rs | 8 +++--- crates/runtime/src/runtime_tests.rs | 20 ++++++++++++--- crates/yellowstone-grpc-source/Cargo.toml | 2 +- crates/yellowstone-grpc-source/src/lib.rs | 31 ++++++++++++++++++++--- 5 files changed, 56 insertions(+), 14 deletions(-) diff --git a/crates/runtime/src/builder.rs b/crates/runtime/src/builder.rs index 43bb429f..bd292672 100644 --- a/crates/runtime/src/builder.rs +++ b/crates/runtime/src/builder.rs @@ -15,9 +15,12 @@ use crate::{ /// Depth of the filter update channel handed to callers by /// [`Runtime::filter_updates`]. Updates are rare, so a shallow queue is enough -/// to keep a caller from blocking on a short burst. The queue does not -/// coalesce: every set sent is forwarded, and the server applies them in order -/// and ends on the newest. +/// to keep a caller from blocking on a short burst. +/// +/// Queued sets are forwarded in order and the server ends on the newest. The +/// exception is a set the sink rejected: the source holds that one and a newer +/// arrival replaces it rather than queueing behind it, so an intermediate set +/// can be skipped after a rejection. const FILTER_UPDATE_CHANNEL_SIZE: usize = 8; /// Helper trait for defining the intended use for a builder. diff --git a/crates/runtime/src/lib.rs b/crates/runtime/src/lib.rs index 9f0a4571..193af83d 100644 --- a/crates/runtime/src/lib.rs +++ b/crates/runtime/src/lib.rs @@ -125,9 +125,11 @@ impl Runtime { /// /// A set the server itself refuses, by exceeding its configured filter /// limits for example, is answered on the stream with a code the client - /// does not treat as recoverable, which ends the run. Sending a set the - /// provider will not accept therefore stops the runtime rather than - /// leaving the previous subscription in place. + /// does not treat as recoverable, and the run ends with that error rather + /// than the previous subscription staying in place. Under [`Self::run`] + /// and [`Self::run_async`] that error is fatal and exits the process, so a + /// set the provider will not accept takes the whole indexer down. Use + /// [`Self::try_run_async`] if a caller needs to survive one. /// /// Returns `None` when the source does not support filter updates, or /// when the sender has already been taken. Call this before running the diff --git a/crates/runtime/src/runtime_tests.rs b/crates/runtime/src/runtime_tests.rs index 322a9bbc..858ac46e 100644 --- a/crates/runtime/src/runtime_tests.rs +++ b/crates/runtime/src/runtime_tests.rs @@ -1,6 +1,6 @@ use std::{ borrow::Cow, - collections::HashMap, + collections::{HashMap, HashSet}, sync::{ atomic::{AtomicUsize, Ordering}, Mutex, @@ -9,7 +9,9 @@ use std::{ }; use async_trait::async_trait; -use shipstern_core::{Filters, ParseResult, Parser, Prefilter, SlotUpdate}; +use shipstern_core::{ + AccountPrefilter, Filters, ParseResult, Parser, Prefilter, Pubkey, SlotUpdate, +}; use tokio::sync::{ mpsc::{Receiver, Sender}, oneshot, @@ -433,10 +435,22 @@ impl SourceTrait for MockFilterUpdateSource { } } +/// A filter set carrying an actual account prefilter. `Prefilter::default()` +/// has every sub-filter unset and converts to an entirely empty +/// `SubscribeRequest`, so a test built on it would pass on a payload that says +/// nothing on the wire. fn filters_for(ids: &[&str]) -> Filters { + let prefilter = Prefilter { + account: Some(AccountPrefilter { + accounts: HashSet::new(), + owners: HashSet::from([Pubkey::default()]), + }), + ..Default::default() + }; + Filters::new( ids.iter() - .map(|id| ((*id).to_owned(), Prefilter::default())) + .map(|id| ((*id).to_owned(), prefilter.clone())) .collect::>(), ) } diff --git a/crates/yellowstone-grpc-source/Cargo.toml b/crates/yellowstone-grpc-source/Cargo.toml index cce7ed42..828908fa 100644 --- a/crates/yellowstone-grpc-source/Cargo.toml +++ b/crates/yellowstone-grpc-source/Cargo.toml @@ -10,7 +10,7 @@ readme = "./README.md" [dependencies] async-trait = { workspace = true } -tokio = { workspace = true, features = ["rt-multi-thread", "signal", "macros"] } +tokio = { workspace = true, features = ["rt-multi-thread", "signal", "macros", "time"] } tracing = { workspace = true } futures-util = { workspace = true, features = ["sink"] } shipstern = { workspace = true } diff --git a/crates/yellowstone-grpc-source/src/lib.rs b/crates/yellowstone-grpc-source/src/lib.rs index 1ffb0185..9515e96e 100644 --- a/crates/yellowstone-grpc-source/src/lib.rs +++ b/crates/yellowstone-grpc-source/src/lib.rs @@ -175,6 +175,9 @@ fn build_subscribe_request(filters: Filters, config: &YellowstoneGrpcConfig) -> request } +/// How often a filter set held after a sink rejection is retried. +const RETRY_HELD_FILTERS_EVERY: Duration = Duration::from_secs(5); + /// Send `filters` to the server, handing it back unsent if the sink rejects it. /// /// A rejection means the request channel is disconnected. With auto-reconnect @@ -196,6 +199,8 @@ async fn send_filter_update( accounts = request.accounts.len(), transactions = request.transactions.len(), slots = request.slots.len(), + blocks = request.blocks.len(), + blocks_meta = request.blocks_meta.len(), "Sending filter update to the live subscription" ); @@ -224,10 +229,7 @@ impl SourceTrait for YellowstoneGrpcSource { tx: Sender>, status_tx: oneshot::Sender, ) -> Result<(), ShipsternError> { - let closed = { - let (_, rx) = mpsc::channel(1); - rx - }; + let (_, closed) = mpsc::channel(1); self.run(tx, status_tx, closed).await } @@ -303,6 +305,13 @@ impl YellowstoneGrpcSource { let mut pending_filters: Option = None; let mut filter_updates_sent: u64 = 0; + // A held set cannot wait on the stream to produce again. The filter it + // is replacing may match nothing, and the client swallows its own + // keepalive messages rather than yielding them, so there are live + // connections on which no update ever arrives to retry from. + let mut retry = tokio::time::interval(RETRY_HELD_FILTERS_EVERY); + retry.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + let exit_status = loop { tokio::select! { update = stream.next() => match update { @@ -348,6 +357,16 @@ impl YellowstoneGrpcSource { }, }, + _ = retry.tick(), if pending_filters.is_some() => { + if let Some(filters) = pending_filters.take() { + pending_filters = send_filter_update(&mut sink, &config, filters).await; + + if pending_filters.is_none() { + filter_updates_sent += 1; + } + } + }, + update = filter_updates.recv(), if accept_filter_updates => { let Some(filters) = update else { // Nobody holds the sending half any more, so leave this @@ -367,6 +386,10 @@ impl YellowstoneGrpcSource { } }; + if pending_filters.is_some() { + tracing::warn!("Connection ended with a filter update still unsent"); + } + let _ = status_tx.send(exit_status); Ok(()) From 62902c3ada3e3df4aa15491b8797c4b0d747675a Mon Sep 17 00:00:00 2001 From: senzenn Date: Tue, 1 Sep 2026 12:23:49 +0530 Subject: [PATCH 9/9] refactor(grpc-source): fold the filter retry into one place Adding the retry timer left the same send-and-count block at three sites and made the copy in the stream arm redundant, since the timer already covers every case. Drop that one so the per-update path carries no filter work at all, and let the send helper own the success count so the two remaining sites are a single line each. Reset the timer when a set is newly held. An interval's first tick is immediate, so without it a rejected set retried at once, inside the same reconnect window that had just rejected it. --- crates/yellowstone-grpc-source/src/lib.rs | 36 ++++++++++------------- 1 file changed, 15 insertions(+), 21 deletions(-) diff --git a/crates/yellowstone-grpc-source/src/lib.rs b/crates/yellowstone-grpc-source/src/lib.rs index 9515e96e..58cd8435 100644 --- a/crates/yellowstone-grpc-source/src/lib.rs +++ b/crates/yellowstone-grpc-source/src/lib.rs @@ -192,6 +192,7 @@ async fn send_filter_update( sink: &mut SubscribeRequestSink, config: &YellowstoneGrpcConfig, filters: Filters, + sent: &mut u64, ) -> Option { let request = build_subscribe_request(filters.clone(), config); @@ -213,6 +214,8 @@ async fn send_filter_update( return Some(filters); } + *sent += 1; + None } @@ -322,18 +325,6 @@ impl YellowstoneGrpcSource { // branch wins first when receiver drops. break SourceExitStatus::ReceiverDropped; } - - // The stream producing again is the signal that any - // reconnect has landed and the sink has a live sender, - // so this is the moment to retry a held filter set. - if let Some(filters) = pending_filters.take() { - pending_filters = - send_filter_update(&mut sink, &config, filters).await; - - if pending_filters.is_none() { - filter_updates_sent += 1; - } - } }, Some(Err(status)) => { // A server that rejects a filter set answers on the stream @@ -359,11 +350,9 @@ impl YellowstoneGrpcSource { _ = retry.tick(), if pending_filters.is_some() => { if let Some(filters) = pending_filters.take() { - pending_filters = send_filter_update(&mut sink, &config, filters).await; - - if pending_filters.is_none() { - filter_updates_sent += 1; - } + pending_filters = + send_filter_update(&mut sink, &config, filters, &mut filter_updates_sent) + .await; } }, @@ -377,10 +366,15 @@ impl YellowstoneGrpcSource { // A newer set supersedes anything still held, because every // set is complete rather than a delta. - pending_filters = send_filter_update(&mut sink, &config, filters).await; - - if pending_filters.is_none() { - filter_updates_sent += 1; + pending_filters = + send_filter_update(&mut sink, &config, filters, &mut filter_updates_sent) + .await; + + // An interval's first tick is immediate, so without this a + // rejected set retries at once, inside the same reconnect + // window that just rejected it. + if pending_filters.is_some() { + retry.reset(); } }, }