fix package spec (#109)

Author: lvboudre

.github/workflows/test.yml

@@ -76,8 +76,8 @@ jobs:
           sudo apt install protobuf-compiler
 
       - name: Run clippy
-        run: cargo clippy --all-targets --tests --features "examples yellowstone-grpc" -- -Dwarnings
+        run: cargo clippy --all-targets --tests --all-features -- -Dwarnings
 
       - name: Test build
         run: |
-          cargo build --features "examples yellowstone-grpc"
+          cargo build --all-features

.gitignore

@@ -2,4 +2,5 @@
 .vscode
 config-gentx-test.yml
 .local
-.env
+.env
+target

Cargo.lock

@@ -3,6 +3,9 @@ members = [
     "apps/jet",
     "crates/tpu-client",
 ]
+exclude = [
+    "examples/jet-tpu-sender",
+]
 resolver = "2"
 
 [workspace.package]
@@ -105,7 +108,7 @@ tonic-prost-build = "0.14.0"
 vergen = "9.0.1"
 tikv-jemallocator = "0.6.1"
 libc = "0.2.177"
-yellowstone-jet-tpu-client = { path = "crates/tpu-client", version = "0.1.0" }
+yellowstone-jet-tpu-client = { path = "crates/tpu-client", version = "0.1.0-rc1" }
 
 [workspace.lints.clippy]
 clone_on_ref_ptr = "deny"

Cargo.toml

@@ -85,7 +85,7 @@ tracing-subscriber = { workspace = true, features = ["ansi", "env-filter", "json
 uuid = { workspace = true, features = ["v4", "serde"] }
 yellowstone-grpc-client = { workspace = true }
 yellowstone-grpc-proto = { workspace = true }
-yellowstone-jet-tpu-client = { workspace = true, features = ["prometheus", "bytes", "yellowstone-grpc"] }
+yellowstone-jet-tpu-client = { workspace = true, features = ["prometheus", "yellowstone-grpc", "shield"] }
 yellowstone-shield-store = { workspace = true }
 tokio-util = { workspace = true }
 tikv-jemallocator = { workspace = true }

apps/jet/Cargo.toml

@@ -348,10 +348,10 @@ async fn run_jet(
         Arc::new(IgnorantLeaderPredictor)
     };
 
+    let (gateway_callback_tx, gateway_response_source) = tokio::sync::mpsc::unbounded_channel();
     let TpuSenderSessionContext {
         identity_updater: gateway_identity_updater,
         driver_tx_sink: gateway_tx_sink,
-        driver_response_source: gateway_response_source,
         driver_join_handle: gateway_join_handle,
     } = quic_gateway_spawner.spawn(
         initial_identity.insecure_clone(),
@@ -360,6 +360,7 @@ async fn run_jet(
             peer_idle_eviction_grace_period: config.quic.connection_idle_eviction_grace,
         }),
         connection_predictor,
+        Some(gateway_callback_tx),
     );
 
     let ah = tg.spawn(async move {

apps/jet/src/bin/jet.rs

@@ -8,7 +8,6 @@ use {
     },
     solana_commitment_config::CommitmentConfig,
     solana_keypair::Keypair,
-    solana_pubkey::Pubkey,
     solana_rpc_client::http_sender::HttpSender,
     std::{
         collections::HashMap,
@@ -43,11 +42,14 @@ use {
         transactions::SendTransactionRequest,
         util::WaitShutdown,
     },
-    yellowstone_jet_tpu_client::yellowstone_grpc::sender::{
-        Blocklist, Endpoints, NewYellowstoneTpuSender, YellowstoneTpuSender,
-        create_yellowstone_tpu_sender,
+    yellowstone_jet_tpu_client::{
+        core::{TpuSenderResponse, TpuSenderResponseCallback},
+        yellowstone_grpc::sender::{
+            Endpoints, NewYellowstoneTpuSender, ShieldBlockList, YellowstoneTpuSender,
+            create_yellowstone_tpu_sender_with_callback,
+        },
     },
-    yellowstone_shield_store::{PolicyStore, PolicyStoreTrait},
+    yellowstone_shield_store::PolicyStore,
 };
 
 #[cfg(not(target_env = "msvc"))]
@@ -194,18 +196,45 @@ async fn run_jet(
         grpc: config.upstream.grpc.endpoint.clone(),
         grpc_x_token: config.upstream.grpc.x_token.clone(),
     };
+    #[derive(Clone)]
+    struct LoggingCallback;
+
+    impl TpuSenderResponseCallback for LoggingCallback {
+        fn call(&self, response: TpuSenderResponse) {
+            use std::io::Write;
+            let mut stdout = std::io::stdout();
+            match response {
+                TpuSenderResponse::TxSent(info) => {
+                    writeln!(
+                        &mut stdout,
+                        "Transaction {} send to {}",
+                        info.tx_sig, info.remote_peer_identity
+                    )
+                    .expect("writeln");
+                }
+                TpuSenderResponse::TxFailed(info) => {
+                    writeln!(&mut stdout, "Transaction failed: {}", info.tx_sig).expect("writeln");
+                }
+                TpuSenderResponse::TxDrop(info) => {
+                    for (txn, _) in info.dropped_tx_vec {
+                        writeln!(&mut stdout, "Transaction dropped: {}", txn.tx_sig)
+                            .expect("writeln");
+                    }
+                }
+            }
+        }
+    }
     let NewYellowstoneTpuSender {
         sender,
         related_objects_jh,
-        response,
-    } = create_yellowstone_tpu_sender(
+    } = create_yellowstone_tpu_sender_with_callback(
         Default::default(),
         initial_identity.insecure_clone(),
         tpu_sender_endpoints,
+        LoggingCallback,
     )
     .await
     .expect("yellowstone-tpu-sender");
-    drop(response); // drop response handle, we don't need it
 
     let ah = tg.spawn(async move {
         related_objects_jh
@@ -225,7 +254,12 @@ async fn run_jet(
         proxy_preflight_check: false,
     };
 
-    let ah = tg.spawn(tpu_sender_loop(scheduler_out, sender, shield_policy_store));
+    let ah = tg.spawn(tpu_sender_loop(
+        scheduler_out,
+        sender,
+        shield_policy_store,
+        jet_cancellation_token.child_token(),
+    ));
 
     tg_name_map.insert(ah.id(), "tpu_sender_loop".to_string());
 
@@ -386,22 +420,12 @@ pub async fn tpu_sender_loop(
     mut incoming: UnboundedReceiver<Arc<SendTransactionRequest>>,
     mut tpu_sender: YellowstoneTpuSender,
     shield: Option<PolicyStore>,
+    cancellation_token: CancellationToken,
 ) {
-    struct ShieldBlocklist<'a> {
-        shield: &'a PolicyStore,
-        policies: &'a [Pubkey],
-    }
-
-    impl Blocklist for ShieldBlocklist<'_> {
-        fn is_blocked(&self, account: &Pubkey) -> bool {
-            self.shield
-                .snapshot()
-                .is_allowed(self.policies, account)
-                .unwrap_or(true)
-        }
-    }
-
     while let Some(request) = incoming.recv().await {
+        if cancellation_token.is_cancelled() {
+            break;
+        }
         let request = Arc::unwrap_or_clone(request);
         let SendTransactionRequest {
             signature,
@@ -411,13 +435,14 @@ pub async fn tpu_sender_loop(
             policies,
         } = request;
 
-        let blocklist = shield.as_ref().map(|shield| ShieldBlocklist {
-            shield,
-            policies: &policies,
+        let blocklist = shield.as_ref().map(|shield| ShieldBlockList {
+            policy_store: shield,
+            shield_policy_addresses: &policies,
+            default_return_value: false,
         });
 
         let result = tpu_sender
-            .send_txn_with_blocklist(signature, wire_transaction, blocklist)
+            .send_txn_fanout_with_blocklist(signature, wire_transaction, 2, blocklist)
             .await;
         if let Err(e) = result {
             tracing::error!(

apps/jet/src/bin/lite-jet.rs

@@ -751,21 +751,10 @@ impl TransactionFanout {
         let lewis_handler = self.lewis_handler.clone();
         let extra_fwd = Arc::clone(&self.extra_fwd);
 
-        enum WireContainer {
-            Bytes(Bytes),
-            SharedVec(Arc<Vec<u8>>),
-        }
-
         let send_fut = async move {
             let next_leaders = leader_schedule_service.leader_lookahead(leader_fwd);
             let current_slot = leader_schedule_service.get_current_slot();
-
-            let txn_wire = if tx.wire_transaction.len() == tx.wire_transaction.capacity() {
-                WireContainer::Bytes(Bytes::from(tx.wire_transaction))
-            } else {
-                WireContainer::SharedVec(Arc::new(tx.wire_transaction))
-            };
-
+            let txn_wire = Bytes::from_owner(tx.wire_transaction);
             for dest in next_leaders.iter() {
                 if !policy_store_service.is_allowed(&tx.policies, dest)? {
                     // Report skip to Lewis
@@ -776,14 +765,7 @@ impl TransactionFanout {
                     tracing::trace!("transaction {signature} is not allowed to be sent to {dest}");
                     continue;
                 }
-                let tpu_txn = match &txn_wire {
-                    WireContainer::Bytes(b) => {
-                        TpuSenderTxn::from_bytes(tx.signature, *dest, b.clone())
-                    }
-                    WireContainer::SharedVec(v) => {
-                        TpuSenderTxn::from_shared_vec(tx.signature, *dest, Arc::clone(v))
-                    }
-                };
+                let tpu_txn = TpuSenderTxn::from_bytes(tx.signature, *dest, txn_wire.clone());
                 tpu_sink
                     .send(tpu_txn)
                     .await
@@ -795,15 +777,7 @@ impl TransactionFanout {
                     // We don't send it twice.
                     continue;
                 }
-                // We don't apply policy here.
-                let tpu_txn = match &txn_wire {
-                    WireContainer::Bytes(b) => {
-                        TpuSenderTxn::from_bytes(tx.signature, *extra, b.clone())
-                    }
-                    WireContainer::SharedVec(v) => {
-                        TpuSenderTxn::from_shared_vec(tx.signature, *extra, Arc::clone(v))
-                    }
-                };
+                let tpu_txn = TpuSenderTxn::from_bytes(tx.signature, *extra, txn_wire.clone());
 
                 tpu_sink
                     .send(tpu_txn)

apps/jet/src/transactions.rs

@@ -6,19 +6,42 @@ authors.workspace = true
 license.workspace = true
 homepage.workspace = true
 repository.workspace = true
+description = "Triton One Yellowstone Jet TPU Client"
+keywords = ["yellowstone", "solana", "rpc", "tpu", "quic"]
 readme = "README.md"
+publish = true
+exclude = ["src/bin/*"]
+
+[[bin]]
+name = "test-managed-schedule"
+path = "src/bin/test-managed-schedule.rs"
+publish = false
+
+[[bin]]
+name = "test-tpu-info"
+path = "src/bin/test-tpu-info.rs"
+publish = false
+
+[[bin]]
+name = "test-tpu-send"
+path = "src/bin/test-tpu-send.rs"
+publish = false
+
+[[bin]]
+name = "test-slot-tracker"
+path = "src/bin/test-slot-tracker.rs"
 publish = false
 
 [features]
-default = ["yellowstone-grpc", "bytes"]
+default = ["yellowstone-grpc"]
 prometheus = ["dep:prometheus"]
-bytes = ["dep:bytes"]
 examples = [
     "dep:dotenvy",
     "dep:tracing-subscriber",
     "dep:clap",
     "dep:solana-system-interface",
     "dep:solana-transaction",
+    "dep:solana-hash",
     "dep:bincode",
     "yellowstone-grpc",
 ]
@@ -35,7 +58,7 @@ base64 = { workspace = true }
 bincode = { workspace = true, optional = true }
 bs58 = { workspace = true }
 borsh = { workspace = true }
-bytes = { workspace = true, optional = true }
+bytes = { workspace = true }
 clap = { workspace = true, features = ["cargo", "derive", "env"], optional = true }
 
 derive_more = { workspace = true, features = ["display"] }
@@ -64,6 +87,7 @@ solana-client = { workspace = true }
 solana-clock = { workspace = true }
 solana-commitment-config = { workspace = true }
 solana-epoch-info = { workspace = true }
+solana-hash = { workspace = true, optional = true }
 solana-instruction = { workspace = true }
 solana-keypair = { workspace = true }
 solana-message = { workspace = true }