feat: Add Jetstreamer source for historical data replay (#149) #151
Conversation
kespinola
left a comment
kespinola
left a comment
There was a problem hiding this comment.
- Run with fmt with
cargo +night fmt --allto remove all the fmt changes - I think we should drop slot ordering to simplify the the source.
- The example should use Vixen runtime
- Make use of the filters in transaction callback
Looking really good though!
|
I have completed the todos and feedback based on the previous PR suggestions you gave me. Kindly review the PR as well, please @kespinola |
| pub(crate) static JETSTREAM_BLOCKS_RECEIVED: LazyLock<IntCounter> = LazyLock::new(|| { | ||
| IntCounter::with_opts(Opts::new( | ||
| "jetstream_blocks_received", | ||
| "Total blocks received from Jetstream", | ||
| )) | ||
| .unwrap() | ||
| }); | ||
|
|
||
| /// Total number of transactions received from Jetstream | ||
| pub(crate) static JETSTREAM_TRANSACTIONS_RECEIVED: LazyLock<IntCounter> = LazyLock::new(|| { | ||
| IntCounter::with_opts(Opts::new( | ||
| "jetstream_transactions_received", | ||
| "Total transactions received from Jetstream", | ||
| )) | ||
| .unwrap() | ||
| }); | ||
|
|
||
| /// Register all Jetstream metrics with the provided Prometheus registry | ||
| pub fn register_metrics(registry: &Registry) { | ||
| let _ = registry.register(Box::new(JETSTREAM_BLOCKS_RECEIVED.clone())); | ||
| let _ = registry.register(Box::new(JETSTREAM_TRANSACTIONS_RECEIVED.clone())); | ||
| } |
There was a problem hiding this comment.
https://github.com/rpcpool/yellowstone-vixen/blob/main/crates/runtime/src/metrics.rs#L250-L279
There is already metric for blocks and transactions received please remove from the source.
| archive_url = "https://api.old-faithful.net" | ||
| epoch = 800 | ||
| threads = 4 | ||
| reorder_buffer_size = 1000 | ||
| slot_timeout_secs = 30 |
There was a problem hiding this comment.
Can you please adjust serde attributes for dash case?
| archive_url = "https://api.old-faithful.net" | |
| epoch = 800 | |
| threads = 4 | |
| reorder_buffer_size = 1000 | |
| slot_timeout_secs = 30 | |
| archive-url = "https://api.old-faithful.net" | |
| epoch = 800 | |
| threads = 4 | |
| reorder-buffer-size = 1000 | |
| slot-timeout-secs = 30 |
| /// Control transaction filtering: true = permissive (all), false = strict (limited). | ||
| #[arg(long, env, default_value = "true")] | ||
| pub permissive_transaction_filtering: bool, |
There was a problem hiding this comment.
Should not have this option its confusing API. You can make a parser that matches all instead.
There was a problem hiding this comment.
removed certain confusing api's in Lts Pr
- Remove duplicate metrics, use runtime metrics - Add serde kebab-case for configs - Update examples to use dash-case fields - Enable prometheus via runtime
…ng from jetstreamer-source
|
@kespinola I’ve resolved the majority of your feedback could you take another look at this PR? |
kespinola
left a comment
kespinola
left a comment
There was a problem hiding this comment.
Last round of CR related to setting the filters key on SubscribeUpdate for matching event to a parser.
| [filters] | ||
| # Add filters for specific programs/accounts if needed | ||
| # programs = ["11111111111111111111111111111112"] | ||
|
|
There was a problem hiding this comment.
The filters should come from the parser's attached filters which mimic grpc subscribe filter notation for selecting account owner or address in accounts list for transactions we shouldn't need to specify filters on the source.
| entry_count, | ||
| } => { | ||
| let update = SubscribeUpdate { | ||
| filters: vec![], |
There was a problem hiding this comment.
@senzenn Vixen uses this filter list to match a subscribe update with a parser so you need to include the filter key for Vixen to match on when it goes through the runtime.
| } | ||
| // TODO: Populate transaction field with protobuf-encoded transaction data | ||
| let update = SubscribeUpdate { | ||
| filters: vec![], |
There was a problem hiding this comment.
Same here that want to use the filters to let Vixen know what parser and handler to take the event through.
| Ok(()) | ||
| } | ||
|
|
||
| fn should_process_transaction(&self, tx: &TransactionData) -> bool { |
There was a problem hiding this comment.
Instead of returning boolean this should return Vec to set on the filters key of the subscribe update. If its empty then you can toss out the event and don't send it to the channel.
| /// Control transaction filtering: true = permissive (all), false = strict (limited). | ||
| #[arg(long, env, default_value = "true")] | ||
| pub permissive_transaction_filtering: bool, |
There was a problem hiding this comment.
| /// Control transaction filtering: true = permissive (all), false = strict (limited). | |
| #[arg(long, env, default_value = "true")] | |
| pub permissive_transaction_filtering: bool, |
I'd like to drop as filters are governed by the registered parsers so as long as there is 1 parser registered then it will match what is desired. I imagine this was helpful for testing through so understand why it was initially included.
| let handler_clone = handler.clone(); | ||
| let on_block = Some(move |_thread_id: usize, block: BlockData| { | ||
| let handler = handler_clone.clone(); | ||
| async move { | ||
| handler | ||
| .process_block(block) | ||
| .await | ||
| .map_err(|e| Box::new(e) as Box<dyn std::error::Error + Send>) | ||
| } | ||
| .boxed() | ||
| }); |
There was a problem hiding this comment.
| let handler_clone = handler.clone(); | |
| let on_block = Some(move |_thread_id: usize, block: BlockData| { | |
| let handler = handler_clone.clone(); | |
| async move { | |
| handler | |
| .process_block(block) | |
| .await | |
| .map_err(|e| Box::new(e) as Box<dyn std::error::Error + Send>) | |
| } | |
| .boxed() | |
| }); | |
| let handler_on_block = handler.clone(); | |
| let on_block = Some(move |_thread_id: usize, block: BlockData| { | |
| let handler_callback = handler_on_block.clone(); | |
| async move { | |
| handler | |
| .process_block(block) | |
| .await | |
| .map_err(|e| Box::new(e) as Box<dyn std::error::Error + Send>) | |
| } | |
| .boxed() | |
| }); |
| //! Prometheus metrics for Jetstream source | ||
| use prometheus::Registry; | ||
|
|
||
| /// Register all Jetstream metrics with the provided Prometheus registry | ||
| pub fn register_metrics(_registry: &Registry) { | ||
| // No metrics to register currently | ||
| } |
| jetstreamer-firehose = "0.1.6" | ||
| jetstreamer-utils = "0.1.6" |
There was a problem hiding this comment.
Can you include in the root Cargo.toml and then reference through the workspace so all versions are set in one place. We can also allow 0.1 to allow matching of any patches that come out.
| yellowstone-vixen-jetstream-source = { path = "../../crates/jetstreamer-source", features = ["prometheus"] } | ||
| yellowstone-vixen-parser = { path = "../../crates/parser", features = ["token-program"] } |
There was a problem hiding this comment.
Can these be referenced by workspace once set there?
| yellowstone-vixen-jetstream-source = { path = "../../crates/jetstreamer-source", features = ["prometheus"] } | ||
| yellowstone-vixen-parser = { path = "../../crates/parser", features = ["token-program"] } | ||
| prometheus = { workspace = true } | ||
| warp = "0.3" |
There was a problem hiding this comment.
Include in Cargo.toml of root and reference through workspace.
3 participants
Summary
Adds Jetstreamer as a new data source for replaying historical Solana ledger data from Old Faithful archive.
Closes #149
Changes
jetstreamer-sourcecrate withSourceTraitimplementationjetstream-replayfor epoch 800Vixen.example.tomlwith Jetstream configUsage
Requirements
jetstreamer-utilsdependency)Acceptance Criteria
Testing
cargo test --package jetstreamer-source cargo run --example jetstream-replay