use base64+zstd account subscription; add dependency on libzstd (or ubuntu package libzstd-dev)

Author: richardbrooks2000

CMakeLists.txt

@@ -61,7 +61,7 @@ set( PC_HDR
 add_library( pc STATIC ${PC_SRC} )
 
 # dependencies
-set( PC_DEP pc ssl crypto z )
+set( PC_DEP pc ssl crypto z zstd )
 
 #
 # applications

README.md

@@ -8,6 +8,9 @@ apt install libssl-dev
 # depends on libz
 apt install zlib1g zlib1g-dev
 
+# depends on libzstd
+apt install libzstd-dev
+
 # uses cmake to build
 apt install cmake
 

pc/manager.cpp

@@ -315,6 +315,8 @@ bool manager::init()
   }
   PC_LOG_INF( "initialized" )
     .add( "version", PC_VERSION )
+    .add( "rpc_host", get_rpc_host() )
+    .add( "tx_host", get_tx_host() )
     .add( "capture_file", get_capture_file() )
     .add( "publish_interval(ms)", get_publish_interval() )
     .end();

pc/request.cpp

@@ -558,7 +558,7 @@ void add_product::submit()
   areq_->set_sender( pkey );
   areq_->set_account( &akey_ );
   areq_->set_owner( gpub );
-  areq_->set_space( sizeof( pc_price_t ) );
+  areq_->set_space( PC_PROD_ACC_SIZE );
   sreq_->set_program( gpub );
   sreq_->set_publish( pkey );
   sreq_->set_account( &akey_ );
@@ -1910,7 +1910,8 @@ void product::update( T *res )
     return;
   }
   pc_prod_t *prod;
-  if ( sizeof( pc_prod_t ) > res->get_data( prod ) ||
+  size_t plen = std::max( sizeof(pc_price_t), (size_t)PC_PROD_ACC_SIZE );
+  if ( sizeof( pc_prod_t ) > res->get_data( prod, plen ) ||
        prod->magic_ != PC_MAGIC ||
        !init_from_account( prod ) ) {
     cptr->set_err_msg( "invalid or corrupt product account" );

pc/rpc_client.cpp

@@ -2,6 +2,7 @@
 #include "bincode.hpp"
 #include <unistd.h>
 #include "log.hpp"
+#include <zstd.h>
 
 using namespace pc;
 
@@ -123,10 +124,20 @@ static void send_transaction( json_wtr& msg, bincode& tx )
 rpc_client::rpc_client()
 : hptr_( nullptr ),
   wptr_( nullptr ),
-  id_( 0UL )
+  id_( 0UL ),
+  cxt_( nullptr )
 {
   hp_.cp_ = this;
   wp_.cp_ = this;
+  cxt_ = ZSTD_createDCtx();
+}
+
+rpc_client::~rpc_client()
+{
+  if ( cxt_ ) {
+    ZSTD_freeDCtx( (ZSTD_DCtx*)cxt_ );
+    cxt_ = nullptr;
+  }
 }
 
 void rpc_client::set_http_conn( net_connect *hptr )
@@ -252,6 +263,19 @@ void rpc_client::remove_notify( rpc_request *rptr )
   }
 }
 
+size_t rpc_client::get_data(
+    const char *dptr, size_t dlen, size_t tlen, char *&ptr )
+{
+  tlen = ZSTD_compressBound( tlen );
+  abuf_.resize( dlen );
+  zbuf_.resize( tlen );
+  ZSTD_DCtx *cxt = (ZSTD_DCtx*)cxt_;
+  dlen = dec_base64( (const uint8_t*)dptr, dlen, (uint8_t*)&abuf_[0] );
+  tlen = ZSTD_decompressDCtx( cxt, &zbuf_[0], tlen, &abuf_[0], dlen );
+  ptr = &zbuf_[0];
+  return tlen;
+}
+
 ///////////////////////////////////////////////////////////////////////////
 // rpc_request
 
@@ -452,7 +476,7 @@ void rpc::get_account_info::request( json_wtr& msg )
   msg.add_key( "params", json_wtr::e_arr );
   msg.add_val( *acc_ );
   msg.add_val( json_wtr::e_obj );
-  msg.add_key( "encoding", "base64" );
+  msg.add_key( "encoding", "base64+zstd" );
   msg.add_key( "commitment", commitment_to_str( cmt_ ) );
   msg.pop();
   msg.pop();
@@ -790,7 +814,7 @@ void rpc::account_subscribe::request( json_wtr& msg )
   msg.add_key( "params", json_wtr::e_arr );
   msg.add_val( *acc_ );
   msg.add_val( json_wtr::e_obj );
-  msg.add_key( "encoding", "base64" );
+  msg.add_key( "encoding", "base64+zstd" );
   msg.add_key( "commitment", commitment_to_str( cmt_ ) );
   msg.pop();
   msg.pop();

pc/rpc_client.hpp

@@ -61,13 +61,13 @@ namespace pc
   commitment str_to_commitment( str );
 
   class rpc_request;
-
   // solana rpc REST API client
   class rpc_client : public error
   {
   public:
 
     rpc_client();
+    ~rpc_client();
 
     // rpc http connection
     void set_http_conn( net_connect * );
@@ -90,8 +90,7 @@ namespace pc
     void remove_notify( rpc_request * );
 
     // decode into buffer and return pointer
-    template<class T>
-    size_t get_data( const char *dptr, size_t dlen, T *&ptr );
+    size_t get_data( const char *dptr, size_t dlen, size_t tlen, char*&ptr);
 
     // reset state
     void reset();
@@ -133,7 +132,9 @@ namespace pc
     id_vec_t     reuse_; // reuse id list
     sub_map_t    smap_;  // subscription map
     acc_buf_t    abuf_;  // account decode buffer
+    acc_buf_t    zbuf_;  // account decompress buffer
     uint64_t     id_;    // next request id
+    void        *cxt_;
   };
 
   // rpc response or subscrption callback
@@ -240,16 +241,6 @@ namespace pc
   /////////////////////////////////////////////////////////////////////////
   // wrappers for various solana rpc requests
 
-  template<class T>
-  size_t rpc_client::get_data( const char *dptr, size_t dlen, T *&ptr )
-  {
-    size_t tlen = enc_base64_len( sizeof( T ) );
-    abuf_.resize( std::max( dlen, tlen ) );
-    dec_base64( (const uint8_t*)dptr, dlen, (uint8_t*)&abuf_[0] );
-    ptr = (T*)&abuf_[0];
-    return abuf_.size();
-  }
-
   namespace rpc
   {
     // get account balance, program data and account meta-data
@@ -266,7 +257,7 @@ namespace pc
       uint64_t get_rent_epoch() const;
       bool     get_is_executable() const;
       void     get_owner( const char *&, size_t& ) const;
-      template<class T> size_t get_data( T *& ) const;
+      template<class T> size_t get_data( T *&, size_t len=sizeof(T) ) const;
 
       get_account_info();
       void request( json_wtr& ) override;
@@ -285,9 +276,13 @@ namespace pc
       commitment  cmt_;
     };
 
-    template<class T> size_t get_account_info::get_data( T *&res ) const
+    template<class T>
+    size_t get_account_info::get_data( T *&res, size_t tlen ) const
     {
-      return get_rpc_client()->get_data( dptr_, dlen_, res );
+      char *ptr;
+      size_t len = get_rpc_client()->get_data( dptr_, dlen_, tlen, ptr );
+      res = (T*)ptr;
+      return len;
     }
 
     // recent block hash and fee schedule
@@ -421,7 +416,7 @@ namespace pc
       // results
       uint64_t get_slot() const;
       uint64_t get_lamports() const;
-      template<class T> size_t get_data( T *& ) const;
+      template<class T> size_t get_data( T *&, size_t len=sizeof(T) ) const;
 
       account_subscribe();
       void request( json_wtr& ) override;
@@ -437,9 +432,13 @@ namespace pc
       commitment  cmt_;
     };
 
-    template<class T> size_t account_subscribe::get_data( T *&res ) const
+    template<class T>
+    size_t account_subscribe::get_data( T *&res, size_t tlen ) const
     {
-      return get_rpc_client()->get_data( dptr_, dlen_, res );
+      char *ptr;
+      size_t len = get_rpc_client()->get_data( dptr_, dlen_, tlen, ptr );
+      res = (T*)ptr;
+      return len;
     }
 
     // transaction to transfer funds between accounts

pc/user.cpp

@@ -191,7 +191,8 @@ void user::parse_upd_price( uint32_t tok, uint32_t itok )
       jw_.add_key( "result", 0UL );
       add_tail( itok );
     } else if ( !sptr->get_is_ready_publish() ) {
-      add_error( itok, PC_JSON_NOT_READY, "not ready to publish" );
+      add_error( itok, PC_JSON_NOT_READY,
+          "not ready to publish - check pyth_tx connection" );
     } else if ( !sptr->has_publisher() ) {
       add_error( itok, PC_JSON_MISSING_PERMS, "missing publish permission" );
     } else if ( sptr->get_is_err() ) {

pctest/test_publish.cpp

@@ -302,11 +302,11 @@ void test_publish::on_response( pc::price_sched *ptr, uint64_t sub_id )
       .end();
     // should work once publisher has been permissioned
   } else if ( !sym->get_is_ready_publish() ) {
-    PC_LOG_WRN( "not ready to publish next price" )
+    PC_LOG_WRN( "not ready to publish next price - check pyth_tx connection")
       .add( "symbol", sym->get_symbol() )
       .add( "price_type", pc::price_type_to_str( sym->get_price_type() ) )
       .end();
-    // could be delay in confirmation - try again next time
+    // likely that pyth_tx not yet connected
   } else if ( sym->get_is_err() ) {
     PC_LOG_WRN( "block-chain error" )
       .add( "symbol", sym->get_symbol() )

pctest/test_publish.py

@@ -125,7 +125,13 @@ async def poll( uri ):
   while True:
     msg = json.loads( await ws.recv() )
 #    print(msg)
-    if 'result' in msg:
+    if 'error' in msg:
+      ts = datetime.datetime.utcnow().isoformat()
+      code = msg['error']['code']
+      emsg = msg['error']['message']
+      print( f'{ts} error code: {code} msg: {emsg}' )
+      sys.exit(1)
+    elif 'result' in msg:
       msgid = msg['id']
       if msgid in allids:
         allids[msgid].parse_reply( msg, allsub )