Add ability to send price updates without pyth_tx.

Author: mass

pc/manager.cpp

@@ -344,6 +344,11 @@ bool manager::get_is_tx_send() const
   return tconn_.get_is_send();
 }
 
+bool manager::get_is_rpc_send() const
+{
+  return hconn_.get_is_send() || wconn_.get_is_send();
+}
+
 bool manager::bootstrap()
 {
   int status = PC_PYTH_RPC_CONNECTED | PC_PYTH_HAS_BLOCK_HASH;
@@ -674,7 +679,7 @@ void manager::on_response( rpc::get_slot *res )
 
   PC_LOG_DBG( "received get_slot" )
     .add( "slot", slot_ )
-    .add( "rount_trip_time(ms)", 1e-6*ack_ts )
+    .add( "round_trip_time(ms)", 1e-6*ack_ts )
     .end();
 
   // submit block hash every N slots

pc/manager.hpp

@@ -169,6 +169,7 @@ namespace pc
     void on_response( rpc::program_subscribe * ) override;
     void set_status( int );
     get_mapping *get_last_mapping() const;
+    bool get_is_rpc_send() const;
 
   private:
 

pc/misc.hpp

@@ -4,7 +4,8 @@
 #include <string>
 
 #define PC_PACKED        __attribute__((__packed__))
-#define PC_UNLIKELY(ARG) __builtin_expect((ARG),1)
+#define PC_LIKELY(ARG)   __builtin_expect((ARG),1)
+#define PC_UNLIKELY(ARG) __builtin_expect((ARG),0)
 #define PC_NSECS_IN_SEC  1000000000L
 #define PC_NSECS_IN_MSEC 1000000L
 

pc/request.cpp

@@ -1838,6 +1838,7 @@ price::price( const pub_key& acc, product *prod )
   areq_->set_account( &apub_ );
   preq_->set_account( &apub_ );
   areq_->set_sub( this );
+  preq_->set_sub( this );
   size_t tlen = ZSTD_compressBound( sizeof(pc_price_t) );
   pptr_ = (pc_price_t*)new char[tlen];
   __builtin_memset( pptr_, 0, tlen );
@@ -1973,7 +1974,14 @@ uint64_t price::get_pub_slot() const
 
 bool price::get_is_ready_publish() const
 {
-  return st_ == e_publish && get_manager()->get_is_tx_connect();
+  if ( st_ != e_publish )
+    return false;
+  manager *cptr = get_manager();
+  if ( cptr->get_do_tx() ) {
+    return cptr->get_is_tx_connect();
+  } else {
+    return cptr->has_status( PC_PYTH_RPC_CONNECTED | PC_PYTH_HAS_BLOCK_HASH );
+  }
 }
 
 void price::reset()
@@ -2033,9 +2041,34 @@ bool price::update(
     return false;
   }
   manager *mgr = get_manager();
-  preq_->set_price( price, conf, st, mgr->get_slot(), is_agg );
+  const uint64_t slot = mgr->get_slot();
+  preq_->set_price( price, conf, st, slot, is_agg );
   preq_->set_block_hash( mgr->get_recent_block_hash() );
-  mgr->submit( preq_ );
+  if ( mgr->get_do_tx() )
+    mgr->submit( preq_ );
+  else {
+    get_rpc_client()->send( preq_ );
+    tvec_.emplace_back( std::string( '\0' , 100 ), preq_->get_sent_time() );
+    preq_->get_signature()->enc_base58( tvec_.back().first );
+    PC_LOG_DBG( "sent price update transaction" )
+      .add( "price_account", *get_account() )
+      .add( "product_account", *prod_->get_account() )
+      .add( "symbol", get_symbol() )
+      .add( "price_type", price_type_to_str( get_price_type() ) )
+      .add( "sig", tvec_.back().first )
+      .add( "pub_slot", slot )
+      .end();
+    if ( PC_UNLIKELY( tvec_.size() >= 100 ) ) {
+      PC_LOG_WRN( "too many unacked price update transactions" )
+        .add( "price_account", *get_account() )
+        .add( "product_account", *prod_->get_account() )
+        .add( "symbol", get_symbol() )
+        .add( "price_type", price_type_to_str( get_price_type() ) )
+        .add( "num_txid", tvec_.size() )
+        .end();
+      tvec_.erase( tvec_.begin(), tvec_.begin() + 50 );
+    }
+  }
   inc_sent();
   return true;
 }
@@ -2051,6 +2084,30 @@ void price::submit()
   }
 }
 
+bool price::has_unacked_updates() const
+{
+  return ! tvec_.empty();
+}
+
+void price::on_response( rpc::upd_price *res )
+{
+  std::string txid = res->get_ack_signature().as_string();
+  const auto it = std::find_if( tvec_.begin(), tvec_.end(),
+      [&] ( const std::pair<std::string,int64_t>& m ) { return m.first == txid; } );
+  if ( it == tvec_.end() )
+    return;
+  const int64_t ack_dur = res->get_recv_time() - it->second;
+  tvec_.erase( it );
+  PC_LOG_DBG( "received price update transaction ack" )
+    .add( "price_account", *get_account() )
+    .add( "product_account", *prod_->get_account() )
+    .add( "symbol", get_symbol() )
+    .add( "price_type", price_type_to_str( get_price_type() ) )
+    .add( "sig", txid )
+    .add( "round_trip_time(ms)", 1e-6 * ack_dur )
+    .end();
+}
+
 void price::on_response( rpc::get_account_info *res )
 {
   set_is_recv( true );

pc/request.hpp

@@ -646,7 +646,8 @@ namespace pc
   // price subscriber and publisher
   class price : public request,
                 public pub_stats,
-                public rpc_sub_i<rpc::get_account_info>
+                public rpc_sub_i<rpc::get_account_info>,
+                public rpc_sub_i<rpc::upd_price>
   {
   public:
 
@@ -680,6 +681,9 @@ namespace pc
     // update aggregate price only
     bool update();
 
+    // are there any update transactions which have not yet been acked
+    bool has_unacked_updates() const;
+
     // get and activate price schedule subscription
     price_sched *get_sched();
 
@@ -730,6 +734,7 @@ namespace pc
     void reset();
     void unsubscribe();
     void submit() override;
+    void on_response( rpc::upd_price * ) override;
     void on_response( rpc::get_account_info * ) override;
     void on_response( rpc::program_subscribe * ) override;
     bool get_is_done() const override;
@@ -739,6 +744,8 @@ namespace pc
     typedef enum {
       e_subscribe, e_sent_subscribe, e_publish, e_error } state_t;
 
+    typedef std::vector<std::pair<std::string,int64_t>> txid_vec_t;
+
     template<class T> void update( T *res );
 
     bool init_publish();
@@ -760,6 +767,7 @@ namespace pc
     rpc::get_account_info  areq_[1];
     rpc::upd_price         preq_[1];
     pc_price_t            *pptr_;
+    txid_vec_t             tvec_;
   };
 
   template<class T>

pc/rpc_client.cpp

@@ -107,13 +107,15 @@ namespace pc
 static hash sys_id = gen_sys_id();
 
 // generate json for sendTransaction
-static void send_transaction( json_wtr& msg, bincode& tx )
+static void send_transaction( json_wtr& msg, bincode& tx, bool skipPreflight )
 {
   msg.add_key( "method", "sendTransaction" );
   msg.add_key( "params", json_wtr::e_arr );
   msg.add_val_enc_base64( str( tx.get_buf(), tx.size() ) );
   msg.add_val( json_wtr::e_obj );
   msg.add_key( "encoding", "base64" );
+  if ( skipPreflight )
+    msg.add_key( "skipPreflight", json_wtr::jtrue() );
   msg.pop();
   msg.pop();
 }
@@ -808,7 +810,7 @@ void rpc::transfer::request( json_wtr& msg )
   sig_.init_from_buf( (const uint8_t*)(tx.get_buf() + sign_idx) );
 
   // encode transaction and add to json params
-  send_transaction( msg, tx );
+  send_transaction( msg, tx, false );
   bptr->dealloc();
 }
 
@@ -1140,7 +1142,7 @@ void rpc::create_account::request( json_wtr& msg )
   tx.sign( acct_idx, tx_idx, *account_ );
 
   // encode transaction and add to json params
-  send_transaction( msg, tx );
+  send_transaction( msg, tx, false );
   bptr->dealloc();
 }
 
@@ -1240,7 +1242,7 @@ void rpc::add_product::request( json_wtr& msg )
   tx.sign( sym_idx, tx_idx, *akey_ );
 
   // encode transaction and add to json params
-  send_transaction( msg, tx );
+  send_transaction( msg, tx, false );
   bptr->dealloc();
 }
 
@@ -1344,7 +1346,7 @@ void rpc::upd_product::request( json_wtr& msg )
   tx.sign( sym_idx, tx_idx, *akey_ );
 
   // encode transaction and add to json params
-  send_transaction( msg, tx );
+  send_transaction( msg, tx, false );
   bptr->dealloc();
 }
 
@@ -1464,7 +1466,7 @@ void rpc::add_price::request( json_wtr& msg )
   tx.sign( prc_idx, tx_idx, *akey_ );
 
   // encode transaction and add to json params
-  send_transaction( msg, tx );
+  send_transaction( msg, tx, false );
   bptr->dealloc();
 }
 
@@ -1573,7 +1575,7 @@ void rpc::init_price::request( json_wtr& msg )
   tx.sign( prc_idx, tx_idx, *akey_ );
 
   // encode transaction and add to json params
-  send_transaction( msg, tx );
+  send_transaction( msg, tx, false );
   bptr->dealloc();
 }
 
@@ -1655,7 +1657,7 @@ void rpc::init_mapping::request( json_wtr& msg )
   tx.sign( map_idx, tx_idx, *mkey_ );
 
   // encode transaction and add to json params
-  send_transaction( msg, tx );
+  send_transaction( msg, tx, false );
   bptr->dealloc();
 }
 
@@ -1746,7 +1748,7 @@ void rpc::add_mapping::request( json_wtr& msg )
   tx.sign( acc_idx, tx_idx, *akey_ );
 
   // encode transaction and add to json params
-  send_transaction( msg, tx );
+  send_transaction( msg, tx, false );
   bptr->dealloc();
 }
 
@@ -1839,7 +1841,7 @@ void rpc::add_publisher::request( json_wtr& msg )
   tx.sign( sym_idx, tx_idx, *akey_ );
 
   // encode transaction and add to json params
-  send_transaction( msg, tx );
+  send_transaction( msg, tx, false );
   bptr->dealloc();
 }
 
@@ -1932,7 +1934,7 @@ void rpc::del_publisher::request( json_wtr& msg )
   tx.sign( sym_idx, tx_idx, *akey_ );
 
   // encode transaction and add to json params
-  send_transaction( msg, tx );
+  send_transaction( msg, tx, false );
   bptr->dealloc();
 }
 
@@ -2014,7 +2016,7 @@ void rpc::init_test::request( json_wtr& msg )
   tx.sign( prm_idx, tx_idx, *akey_ );
 
   // encode transaction and add to json params
-  send_transaction( msg, tx );
+  send_transaction( msg, tx, false );
   bptr->dealloc();
 }
 
@@ -2122,16 +2124,7 @@ void rpc::upd_test::request( json_wtr& msg )
   tx.sign( tst_idx, tx_idx, *tkey_ );
 
   // encode transaction and add to json params
-  msg.add_key( "method", "sendTransaction" );
-  msg.add_key( "params", json_wtr::e_arr );
-  char buf[4096];
-  size_t buf_len = enc_base64( (const uint8_t*)tx.get_buf(),
-      tx.size(), (uint8_t*)buf );
-  msg.add_val( str( buf, buf_len ) );
-  msg.add_val( json_wtr::e_obj );
-  msg.add_key( "encoding", "base64" );
-  msg.pop();
-  msg.pop();
+  send_transaction( msg, tx, false );
   bptr->dealloc();
 }
 
@@ -2314,6 +2307,16 @@ void rpc::upd_price::set_price( int64_t px,
   cmd_   = is_agg?e_cmd_agg_price:e_cmd_upd_price;
 }
 
+signature *rpc::upd_price::get_signature()
+{
+  return &sig_;
+}
+
+str rpc::upd_price::get_ack_signature() const
+{
+  return ack_sig_;
+}
+
 class tx_wtr : public net_wtr
 {
 public:
@@ -2329,12 +2332,8 @@ class tx_wtr : public net_wtr
   }
 };
 
-void rpc::upd_price::build( net_wtr& wtr )
+void rpc::upd_price::build_tx( bincode& tx )
 {
-  // construct binary transaction and add header
-  bincode tx;
-  ((tx_wtr&)wtr).init( tx );
-
   // signatures section
   tx.add_len<1>();      // one signature (publish)
   size_t pub_idx = tx.reserve_sign();
@@ -2376,5 +2375,36 @@ void rpc::upd_price::build( net_wtr& wtr )
 
   // all accounts need to sign transaction
   tx.sign( pub_idx, tx_idx, *ckey_ );
+  sig_.init_from_buf( (const uint8_t*)(tx.get_buf() + pub_idx) );
+}
+
+void rpc::upd_price::build( net_wtr& wtr )
+{
+  bincode tx;
+  ((tx_wtr&)wtr).init( tx );
+  build_tx( tx );
   ((tx_wtr&)wtr).commit( tx );
 }
+
+void rpc::upd_price::request( json_wtr& msg )
+{
+  // construct binary transaction
+  net_buf *bptr = net_buf::alloc();
+  bincode tx( bptr->buf_ );
+  build_tx( tx );
+
+  // encode transaction and add to json params
+  send_transaction( msg, tx, true );
+  bptr->dealloc();
+}
+
+void rpc::upd_price::response( const jtree& jt )
+{
+  if ( on_error( jt, this ) )
+    return;
+  uint32_t rtok = jt.find_val( 1, "result" );
+  if ( rtok == 0 )
+    return;
+  ack_sig_ = jt.get_str( rtok );
+  on_response( this );
+}