Support for pyth_tx proxy server for forwarding transactions directly to tpu ports on solana nodes

Author: richardbrooks2000

CMakeLists.txt

@@ -73,6 +73,8 @@ add_executable( pyth pcapps/pyth.cpp )
 target_link_libraries( pyth ${PC_DEP} )
 add_executable( pyth_csv pcapps/pyth_csv.cpp )
 target_link_libraries( pyth_csv ${PC_DEP} )
+add_executable( pyth_tx pcapps/tx_svr.cpp pcapps/pyth_tx.cpp )
+target_link_libraries( pyth_tx ${PC_DEP} )
 
 #
 # install

dashboard/dashboard.js

@@ -74,10 +74,14 @@ class Prices
     let tab = document.getElementById( "prices" );
     let k = 1;
     res.result.sort( function(a,b) {
-      return a['attr_dict']['symbol'] > b['attr_dict']['symbol'];
+      let asym = a['attr_dict']['symbol'];
+      let bsym = b['attr_dict']['symbol'];
+      return ( asym == bsym ? 0 : (asym > bsym ? 1: -1 ) );
     } );
     res.result.sort( function(a,b) {
-      return a['attr_dict']['asset_type'] > b['attr_dict']['asset_type'];
+      let atype = a['attr_dict']['asset_type'];
+      let btype = b['attr_dict']['asset_type'];
+      return ( atype == btype ? 0 : (atype > btype ? 1: -1 ) );
     } );
     for( let i = 0; i != res.result.length; ++i ) {
       let sym = res.result[i];

doc/getting_started.md

@@ -47,16 +47,16 @@ Once permissioned, you can test your setup by running the test_publish.cpp examp
 
 
 ```
-KHOST=44.232.27.44 # or devnet.solana.com
-./test_publish -k $KDIR -r $KHOST
+KHOST=44.232.27.44 # or api.devnet.solana.com
+./test_publish -k $KDIR -r $KHOST -t $KHOST
 ```
 
 The certification environment (pythnet) can be found at the IP address 44.232.27.44. Please also provide to the administrator the IP address you plan to publish to pythnet from so that it can be added to the pythnet firewall.  Solana devnet can be found at: devnet.solana.com. No additional permissioning is required to publish to devnet.
 
 You can also publish to solana using the pythd server. Start up the server using the same key-store directory and host specification:
 
 ```
-./pythd -k $KDIR -r $KHOST
+./pythd -k $KDIR -r $KHOST -t $KHOST
 ```
 
 Run the test_publish.py example program on the same host to connect to the pythd server:

doc/publish_slot.md

@@ -0,0 +1,54 @@
+# Understanding publishing slots
+
+When a quoter publishes a price, the pyth-client API also forwards what it thinks is the current slot on solana. This is known as its publishing slot.
+
+The publishing slot and price is stored as the latest update for that publisher on-chain but only if the price is for a later slot than that currently stored. This is to prevent prices from being updated out-of-order and to facilitate arbitration between multiple publishers.
+
+The aggregation algorithm only combines prices from publishers that were published within 8 slots of the current on-chain slot.
+
+Not all published prices get included in the pyth contract due to unreliable transports and the way solana formulates and reaches consensus on each slot.
+
+A quoter may detect if a published price is dropped by comparing the list of publishing slots it submits vs what it subsequently receives in each aggregate price callback.
+
+For example, here is an excerpt of a log take from a run of the test_publish.cpp example program against mainnet-beta. It logs everything it sends and everything it receives.
+
+The publishing slots of six consecutive price submissions have been annotated with the labels A, B, C, D, E and F or slots 79018079, 79018084, 79018085, 79018086, 79018087, 79018092.
+
+The API submits a new price every time it receives notification of a new slot but note that prices for slots 79018080 thru 79018083 and 79018088 thru 79018091 were not submitted. This is because solana does not always publish consecutive slots and gaps can occur.  Solana can also publish slots out-of-order, but the API ignores these and is guaranteed only to issue callbacks for slots that are strictly increasing.
+
+Price updates occur for slots labelled A, B, C and F. Slots D and E (79018086, 79018087) were dropped and did not get executed on the chain.
+
+```
+[2021-05-18T22:36:14.048435Z 654359 INF submit price to block-chain             ] symbol=SYMBOL1/USD,price_type=price,price=0.116000,spread=0.001000,slot=79018079,sub_id=1
+                                                                                                                                                          ^^ A ^^^
+[2021-05-18T22:36:14.237644Z 654359 INF received aggregate price update         ] symbol=SYMBOL1/USD,price_type=price,status=trading,agg_price=0.112000,agg_spread=0.001000,valid_slot=79018076,pub_slot=79018077,my_price=0.112000,my_conf=0.001000,my_status=trading,my_slot=79018075
+[2021-05-18T22:36:14.405182Z 654359 INF received aggregate price update         ] symbol=SYMBOL1/USD,price_type=price,status=trading,agg_price=0.113000,agg_spread=0.001000,valid_slot=79018077,pub_slot=79018078,my_price=0.113000,my_conf=0.001000,my_status=trading,my_slot=79018076
+[2021-05-18T22:36:16.099126Z 654359 INF submit price to block-chain             ] symbol=SYMBOL1/USD,price_type=price,price=0.117000,spread=0.001000,slot=79018084,sub_id=1
+                                                                                                                                                          ^^ B ^^^
+[2021-05-18T22:36:16.962077Z 654359 INF submit price to block-chain             ] symbol=SYMBOL1/USD,price_type=price,price=0.118000,spread=0.001000,slot=79018085,sub_id=1
+                                                                                                                                                          ^^ C ^^^
+[2021-05-18T22:36:17.519741Z 654359 INF submit price to block-chain             ] symbol=SYMBOL1/USD,price_type=price,price=0.119000,spread=0.001000,slot=79018086,sub_id=1
+                                                                                                                                                          ^^ D ^^^
+[2021-05-18T22:36:17.671924Z 654359 INF received aggregate price update         ] symbol=SYMBOL1/USD,price_type=price,status=trading,agg_price=0.114000,agg_spread=0.001000,valid_slot=79018078,pub_slot=79018079,my_price=0.114000,my_conf=0.001000,my_status=trading,my_slot=79018077
+[2021-05-18T22:36:18.109491Z 654359 INF submit price to block-chain             ] symbol=SYMBOL1/USD,price_type=price,price=0.120000,spread=0.001000,slot=79018087,sub_id=1
+                                                                                                                                                          ^^ E ^^^
+[2021-05-18T22:36:20.537479Z 654359 INF submit price to block-chain             ] symbol=SYMBOL1/USD,price_type=price,price=0.121000,spread=0.001000,slot=79018092,sub_id=1
+                                                                                                                                                          ^^ F ^^^
+[2021-05-18T22:36:21.195836Z 654359 INF submit price to block-chain             ] symbol=SYMBOL1/USD,price_type=price,price=0.122000,spread=0.001000,slot=79018093,sub_id=1
+[2021-05-18T22:36:21.529074Z 654359 INF submit price to block-chain             ] symbol=SYMBOL1/USD,price_type=price,price=0.123000,spread=0.001000,slot=79018094,sub_id=1
+[2021-05-18T22:36:21.802004Z 654359 INF received aggregate price update         ] symbol=SYMBOL1/USD,price_type=price,status=trading,agg_price=0.116000,agg_spread=0.001000,valid_slot=79018079,pub_slot=79018085,my_price=0.116000,my_conf=0.001000,my_status=trading,my_slot=79018079
+                                                                                                                                                                                                                                                                               ^^ A ^^^
+[2021-05-18T22:36:21.969477Z 654359 INF received aggregate price update         ] symbol=SYMBOL1/USD,price_type=price,status=trading,agg_price=0.117000,agg_spread=0.001000,valid_slot=79018085,pub_slot=79018087,my_price=0.117000,my_conf=0.001000,my_status=trading,my_slot=79018084
+                                                                                                                                                                                                                                                                               ^^ B ^^^
+[2021-05-18T22:36:22.304469Z 654359 INF received aggregate price update         ] symbol=SYMBOL1/USD,price_type=price,status=trading,agg_price=0.118000,agg_spread=0.001000,valid_slot=79018087,pub_slot=79018093,my_price=0.118000,my_conf=0.001000,my_status=trading,my_slot=79018085
+                                                                                                                                                                                                                                                                               ^^ C ^^^
+[2021-05-18T22:36:22.758348Z 654359 INF submit price to block-chain             ] symbol=SYMBOL1/USD,price_type=price,price=0.125000,spread=0.001000,slot=79018096,sub_id=1
+[2021-05-18T22:36:23.121339Z 654359 INF received aggregate price update         ] symbol=SYMBOL1/USD,price_type=price,status=trading,agg_price=0.121000,agg_spread=0.001000,valid_slot=79018093,pub_slot=79018094,my_price=0.121000,my_conf=0.001000,my_status=trading,my_slot=79018092
+                                                                                                                                                                                                                                                                               ^^ F ^^^
+```
+
+The API keeps track of the "hit-rate" of price submissions that show up in the update callbacks and tracks end-to-end latency statistics at the 25th, 50th, 75th and 99th percentiles both in terms of seconds of elapsed time and in number of slot updates observed. For example, from the same log:
+
+```
+[2021-05-18T22:37:26.685518Z 654359 INF publish statistics                      ] symbol=SYMBOL1/USD,price_type=price,num_sent=135,hit_rate=73.333333,secs_p25=2.000000,secs_p50=2.500000,secs_p75=3.000000,secs_p99=7.500000,slot_p25=4,slot_p50=4,slot_p75=6,slot_p99=16
+```

pc/bincode.hpp

@@ -33,6 +33,7 @@ namespace pc
 
     // add values to buffer
     void add( uint8_t );
+    void add( uint16_t );
     void add( uint32_t );
     void add( uint64_t );
     void add( int32_t );
@@ -117,6 +118,11 @@ namespace pc
     idx_ += sizeof( T );
   }
 
+  inline void bincode::add( uint16_t val )
+  {
+    add_val_T( val );
+  }
+
   inline void bincode::add( uint32_t val )
   {
     add_val_T( val );

pc/manager.cpp

@@ -3,11 +3,12 @@
 
 using namespace pc;
 
+#define PC_TPU_PROXY_PORT     8898
 #define PC_RPC_HTTP_PORT      8899
-#define PC_RPC_WEBSOCKET_PORT 8900
 #define PC_RECONNECT_TIMEOUT  (120L*1000000000L)
 #define PC_BLOCKHASH_TIMEOUT  3
 #define PC_PUB_INTERVAL       (293L*PC_NSECS_IN_MSEC)
+#define PC_RPC_HOST           "localhost"
 
 ///////////////////////////////////////////////////////////////////////////
 // manager_sub
@@ -24,6 +25,14 @@ void manager_sub::on_disconnect( manager * )
 {
 }
 
+void manager_sub::on_tx_connect( manager * )
+{
+}
+
+void manager_sub::on_tx_disconnect( manager * )
+{
+}
+
 void manager_sub::on_init( manager * )
 {
 }
@@ -36,7 +45,9 @@ void manager_sub::on_add_symbol( manager *, price * )
 // manager
 
 manager::manager()
-: sub_( nullptr ),
+: thost_( PC_RPC_HOST ),
+  rhost_( PC_RPC_HOST ),
+  sub_( nullptr ),
   status_( 0 ),
   num_sub_( 0 ),
   kidx_( (unsigned)-1 ),
@@ -49,14 +60,17 @@ manager::manager()
   pub_int_( PC_PUB_INTERVAL ),
   wait_conn_( false ),
   do_cap_( false ),
+  do_tx_( true ),
   is_pub_( false )
 {
+  tconn_.set_sub( this );
   breq_->set_sub( this );
   sreq_->set_sub( this );
 }
 
 manager::~manager()
 {
+  teardown();
   for( get_mapping *mptr: mvec_ ) {
     delete mptr;
   }
@@ -65,7 +79,6 @@ manager::~manager()
     delete ptr;
   }
   svec_.clear();
-  teardown();
 }
 
 void manager::add_map_sub()
@@ -95,6 +108,26 @@ std::string manager::get_rpc_host() const
   return rhost_;
 }
 
+void manager::set_tx_host( const std::string& thost )
+{
+  thost_ = thost;
+}
+
+std::string manager::get_tx_host() const
+{
+  return thost_;
+}
+
+void manager::set_do_tx( bool do_tx )
+{
+  do_tx_ = do_tx;
+}
+
+bool manager::get_do_tx() const
+{
+  return do_tx_;
+}
+
 void manager::set_capture_file( const std::string& cap_file )
 {
   cap_.set_file( cap_file );
@@ -226,17 +259,38 @@ bool manager::init()
     return set_err_msg( nl_.get_err_msg() );
   }
 
+  // decompose rpc_host into host:port
+  int rport =0, wport = 0;
+  std::string rhost = get_host_port( rhost_, rport, wport );
+  if ( rport == 0 ) rport = PC_RPC_HTTP_PORT;
+  if ( wport == 0 ) wport = rport+1;
+
   // add rpc_client connection to net_loop and initialize
-  hconn_.set_port( PC_RPC_HTTP_PORT );
-  hconn_.set_host( rhost_ );
+  hconn_.set_port( rport );
+  hconn_.set_host( rhost );
   hconn_.set_net_loop( &nl_ );
   clnt_.set_http_conn( &hconn_ );
-  wconn_.set_port( PC_RPC_WEBSOCKET_PORT );
-  wconn_.set_host( rhost_ );
+  wconn_.set_port( wport );
+  wconn_.set_host( rhost );
   wconn_.set_net_loop( &nl_ );
   clnt_.set_ws_conn( &wconn_ );
-  hconn_.init();
-  wconn_.init();
+  if ( !hconn_.init() ) {
+    return set_err_msg( hconn_.get_err_msg() );
+  }
+  if ( !wconn_.init() ) {
+    return set_err_msg( wconn_.get_err_msg() );
+  }
+  // connect to pyth_tx server
+  if ( do_tx_ ) {
+    int tport1 = 0, tport2 = 0;
+    std::string thost = get_host_port( thost_, tport1, tport2 );
+    tconn_.set_port( tport1 ? tport1 : PC_TPU_PROXY_PORT );
+    tconn_.set_host( thost );
+    tconn_.set_net_loop( &nl_ );
+    if ( !tconn_.init() ) {
+      return set_err_msg( tconn_.get_err_msg() );
+    }
+  }
   wait_conn_ = true;
 
   // initialize listening port if port defined
@@ -320,6 +374,9 @@ void manager::poll( bool do_wait )
       hconn_.poll();
       wconn_.poll();
     }
+    if ( do_tx_ ) {
+      tconn_.poll();
+    }
     if ( lsvr_.get_port() ) {
       lsvr_.poll();
       for( user *uptr = olist_.first(); uptr; ) {
@@ -346,6 +403,11 @@ void manager::poll( bool do_wait )
   // get current time
   curr_ts_ = get_now();
 
+  // try to (re)connect to tx proxy
+  if ( do_tx_ && ( !tconn_.get_is_connect() || tconn_.get_is_err() ) ) {
+    tconn_.reconnect();
+  }
+
   // submit new quotes while connected
   if ( has_status( PC_PYTH_RPC_CONNECTED ) &&
        !hconn_.get_is_err() &&
@@ -489,6 +551,7 @@ void manager::teardown_users()
     PC_LOG_DBG( "delete_user" ).add("fd", usr->get_fd() ).end();
     usr->close();
     dlist_.del( usr );
+    delete usr;
   }
 }
 
@@ -584,6 +647,31 @@ void manager::submit( request *req )
   plist_.add( req );
 }
 
+void manager::submit( tx_request *req )
+{
+  net_wtr msg;
+  req->build( msg );
+  tconn_.add_send( msg );
+}
+
+void manager::on_connect()
+{
+  // callback user with connection status
+  PC_LOG_INF( "pyth_tx_connected" ).end();
+  if ( sub_ ) {
+    sub_->on_tx_connect( this );
+  }
+}
+
+void manager::on_disconnect()
+{
+  // callback user with connection status
+  PC_LOG_INF( "pyth_tx_reset" ).end();
+  if ( sub_ ) {
+    sub_->on_tx_disconnect( this );
+  }
+}
+
 void manager::add_product( const pub_key&acc )
 {
   acc_map_t::iter_t it = amap_.find( acc );