diff --git a/crates/runtime/src/builder.rs b/crates/runtime/src/builder.rs index a2419798..bd292672 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,16 @@ 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. +/// +/// 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. pub trait BuilderKind: Default { /// The type of error returned by the builder. @@ -265,10 +276,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..193af83d 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,55 @@ 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. + /// + /// 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. + /// + /// 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, 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 + /// 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 +304,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/runtime_tests.rs b/crates/runtime/src/runtime_tests.rs index 381848ce..858ac46e 100644 --- a/crates/runtime/src/runtime_tests.rs +++ b/crates/runtime/src/runtime_tests.rs @@ -1,12 +1,21 @@ use std::{ borrow::Cow, - sync::atomic::{AtomicUsize, Ordering}, + collections::{HashMap, HashSet}, + 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::{ + AccountPrefilter, Filters, ParseResult, Parser, Prefilter, Pubkey, SlotUpdate, +}; +use tokio::sync::{ + mpsc::{Receiver, Sender}, + oneshot, +}; use yellowstone_grpc_proto::{ geyser::{subscribe_update::UpdateOneof, SlotStatus, SubscribeUpdate, SubscribeUpdateSlot}, tonic, @@ -373,6 +382,131 @@ 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(()) + } +} + +/// 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.clone())) + .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() 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 + } } diff --git a/crates/yellowstone-grpc-source/Cargo.toml b/crates/yellowstone-grpc-source/Cargo.toml index b3638a1a..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"] } +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/README.md b/crates/yellowstone-grpc-source/README.md index 8954968d..2a55a747 100644 --- a/crates/yellowstone-grpc-source/README.md +++ b/crates/yellowstone-grpc-source/README.md @@ -19,6 +19,60 @@ 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. + +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 +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 +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: diff --git a/crates/yellowstone-grpc-source/src/lib.rs b/crates/yellowstone-grpc-source/src/lib.rs index 3003fbe5..58cd8435 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}, @@ -149,16 +152,115 @@ 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, 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(); + + if let Some(commitment_level) = config.commitment_level { + request.commitment = Some(commitment_level as i32); + } + + 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 +/// 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, + sent: &mut u64, +) -> Option { + let request = build_subscribe_request(filters.clone(), config); + + tracing::debug!( + 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" + ); + + 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); + } + + *sent += 1; + + 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) = mpsc::channel(1); + + 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(); @@ -179,13 +281,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(), @@ -199,7 +296,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?; @@ -207,29 +304,86 @@ impl SourceTrait for YellowstoneGrpcSource { tracing::debug!("gRPC stream started"); + let mut accept_filter_updates = true; + 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 { - 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; - } + 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; + } + }, + Some(Err(status)) => { + // 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; + break SourceExitStatus::StreamError { code, message }; + }, + None => { + break SourceExitStatus::StreamEnded; + }, }, - 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 }; + + _ = retry.tick(), if pending_filters.is_some() => { + if let Some(filters) = pending_filters.take() { + pending_filters = + send_filter_update(&mut sink, &config, filters, &mut filter_updates_sent) + .await; + } }, - 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, &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(); + } }, } }; + if pending_filters.is_some() { + tracing::warn!("Connection ended with a filter update still unsent"); + } + let _ = status_tx.send(exit_status); Ok(()) @@ -238,7 +392,68 @@ 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 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( + 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` 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( + 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`