import { Market } from '@project-serum/serum' ; import { AccountInfo , Commitment , PublicKey } from '@solana/web3.js' ; import AbortController from 'abort-controller' ; import fetch from 'node-fetch' ; import { PassThrough } from 'stream' ; import WebSocket from 'ws' ; import { executeAndRetry , wait } from './helpers' ; import { logger } from './logger' ; export class RPCClient { constructor ( private readonly _options ) { } public async * streamAccountsNotification ( market , marketName ) { const wsEndpoint = new URL ( this . _options . nodeEndpoint ) ; wsEndpoint . protocol = this . _options . nodeEndpoint . startsWith ( 'https' ) ? 'wss' : 'ws' ; if ( this . _options . wsEndpointPort !== undefined ) { wsEndpoint . port = this . _options . wsEndpointPort . toString ( ) ; } const notificationsStream = new PassThrough ( { objectMode : true , highWaterMark : 8096 } ) ; const options = { nodeWsEndpoint : wsEndpoint . toString ( ) , nodeRestEndpoint : this . _options . nodeEndpoint , marketName , commitment : this . _options . commitment } ; const accountsChangeNotifications = new AccountsChangeNotifications ( market , options ) ; logger . log ( 'info' , 'Starting RPC client' , options ) ; accountsChangeNotifications . onAccountsChange = ( notification ) => { notificationsStream . write ( notification ) ; } ; try { for await ( const notification of notificationsStream ) { yield notification as AccountsNotification ; } } finally { accountsChangeNotifications . dispose ( ) ; } } async getAccountInfo ( publicKey , commitment ? ) { const { result } = await executeAndRetry ( async ( ) => this . _getAccountInfoRPCResponseRaw ( publicKey . toBase58 ( ) , commitment ) , { maxRetries : 10 } ) ; if ( result . value === null ) { return null ; } const accountInfo = { owner : new PublicKey ( result . value . owner ) , data : Buffer . from ( result . value . data [ 0 ] , 'base64' ) , lamports : result . value . lamports , executable : result . value . executable } ; return accountInfo ; } private async _getAccountInfoRPCResponseRaw ( publicKey , commitment ? ) { const controller = new AbortController ( ) ; const requestTimeout = setTimeout ( ( ) => { controller . abort ( ) ; } , 3000 ) ; try { const response = await fetch ( this . _options . nodeEndpoint , { signal : controller . signal , method : 'POST' , headers : { 'Content-Type' : 'application/json' } , body : JSON . stringify ( { jsonrpc : '2.0' , id : 1 , method : 'getAccountInfo' , params : [ publicKey , { encoding : 'base64' , commitment } ] } ) } ) ; if ( ! response . ok ) { let errorText = '' ; try { errorText += await response . text ( ) ; } catch { } throw new Error ( errorText ) ; } const data = ( await response . json ( ) ) as { error : any ; result : { context : { slot : number ; } ; value : { data : [ string , string ] ; executable : boolean ; lamports : number ; owner : string ; rentEpoch : number ; } | null ; } ; } ; if ( data . error !== undefined ) { throw new Error ( `JSON RPC error: ${ JSON . stringify ( data . error ) }` ) ; } return data ; } finally { clearTimeout ( requestTimeout ) ; } } } class AccountsChangeNotifications { private _currentSlot : number | undefined = undefined ; private _state : 'PRISTINE' | 'PENDING' | 'PUBLISHED' = 'PRISTINE' ; private _accountsData : AccountsData = { asks : undefined , bids : undefined , eventQueue : undefined } ; private _slotStartTimestamp : number | undefined = undefined ; private _publishTID : NodeJS . Timer | undefined = undefined ; private _pingTID : NodeJS . Timer | undefined = undefined ; private _staleConnectionTID : NodeJS . Timer | undefined = undefined ; private _retriesCount = 0 ; private _receivedMessagesCount = 0 ; public onAccountsChange : ( notification : AccountsNotification ) => void = ( ) => { } ; private _disposed : boolean = false ; private readonly _accountsMeta : { readonly name : AccountName ; readonly reqId : number ; readonly address : string ; } [ ] ; private _wsSubsMeta : Map < number , AccountName > = new Map ( ) ; constructor ( market , private readonly _options ) { this . _accountsMeta = [ { name : 'bids' , reqId : 1000 , address : market . bidsAddress . toBase58 ( ) } , { name : 'asks' , reqId : 2000 , address : market . asksAddress . toBase58 ( ) } , { name : 'eventQueue' , reqId : 3000 , address : ( market as any ) . _decoded . eventQueue . toBase58 ( ) } ] ; this . _connectAndStreamData ( ) ; } public dispose ( ) { this . _clearTimers ( ) ; this . _disposed = true ; } private _connectAndStreamData ( ) { if ( this . _disposed ) { return ; } const ws = new WebSocket ( this . _options . nodeWsEndpoint , { handshakeTimeout : 15 * 1000 , skipUTF8Validation : true } as any ) ; ws . onopen = async ( ) => { try { this . _subscribeToHeartbeat ( ws ) ; this . _sendPeriodicPings ( ws ) ; this . _monitorConnectionIfStale ( ws ) ; const { accountsData , slot } = await executeAndRetry ( async ( ) => this . _fetchAccountsSnapshot ( ) , { maxRetries : 10 } ) ; this . onAccountsChange ( { accountsData , slot } ) ; this . _currentSlot = slot ; } catch ( err ) { logger . log ( 'warn' , `Failed to fetch accounts snapshot, ${ err . message }` , { market : this . _options . marketName } ) ; } this . _subscribeToAccountsNotifications ( ws ) ; logger . log ( 'info' , 'Established new RPC WebSocket connection...' , { market : this . _options . marketName } ) ; } ; ws . onmessage = ( event ) => { if ( this . _disposed ) { return ; } const message = JSON . parse ( event . data as any ) ; if ( message . error !== undefined ) { logger . log ( 'warn' , `Received RPC WebSocket error message: ${ event . data }` , { market : this . _options . marketName } ) ; ws . terminate ( ) ; return ; } this . _receivedMessagesCount ++ ; if ( message . result !== undefined ) { const matchingAccount = this . _accountsMeta . find ( ( a ) => a . reqId === message . id ) ; if ( matchingAccount !== undefined ) { this . _wsSubsMeta . set ( message . result , matchingAccount . name ) ; } return ; } if ( message . method === 'accountNotification' ) { const subMessage = message as { method : 'accountNotification' ; params : { result : { context : { slot : number ; } ; value : { data : [ string , string ] ; } ; } ; subscription : number ; } ; } ; const subId = subMessage . params . subscription ; const matchingSubMeta = this . _wsSubsMeta . get ( subId ) ; if ( matchingSubMeta !== undefined ) { const accountData = Buffer . from ( subMessage . params . result . value . data [ 0 ] , 'base64' ) ; const slot = message . params . result . context . slot ; this . _update ( matchingSubMeta , accountData , slot ) ; } else { throw new Error ( `Unknown notification message (no matching sub id)` ) ; } return ; } if ( message . method === 'slotNotification' ) { return ; } throw new Error ( `Unknown message ` + message . method ) ; } ; ws . onerror = ( event ) => { logger . log ( 'info' , `Received RPC WebSocket error, error message: ${ event . message }` , { market : this . _options . marketName } ) ; } ; ws . onclose = ( event ) => { logger . log ( 'info' , `Received RPC WebSocket close, reason: ${ event . reason }, code: ${ event . code }` , { market : this . _options . marketName } ) ; this . _restartConnection ( ) ; } ; } private async _fetchAccountsSnapshot ( ) { const controller = new AbortController ( ) ; const requestTimeout = setTimeout ( ( ) => { controller . abort ( ) ; } , 5000 ) ; try { const response = await fetch ( this . _options . nodeRestEndpoint , { signal : controller . signal , method : 'POST' , headers : { 'Content-Type' : 'application/json' } , body : JSON . stringify ( { jsonrpc : '2.0' , id : 1 , method : 'getMultipleAccounts' , params : [ this . _accountsMeta . map ( ( a ) => a . address ) , { encoding : 'base64' , commitment : this . _options . commitment } ] } ) } ) ; if ( ! response . ok ) { let errorText = '' ; try { errorText += await response . text ( ) ; } catch { } throw new Error ( errorText ) ; } const data = ( await response . json ( ) ) as { error : any ; result : { context : { slot : number ; } ; value : { data : [ string , 'base64' ] ; } [ ] ; } ; } ; if ( data . error !== undefined ) { throw new Error ( `JSON RPC error: ${ JSON . stringify ( data . error ) }` ) ; } const accountsData = { } ; for ( let i = 0 ; i < data . result . value . length ; i ++ ) { const accountName = this . _accountsMeta [ i ] ! . name ; accountsData [ accountName ] = Buffer . from ( data . result . value [ i ] ! . data [ 0 ] , 'base64' ) ; } return { accountsData , slot : data . result . context . slot } ; } finally { clearTimeout ( requestTimeout ) ; } } private async _restartConnection ( ) { const delayMs = this . _retriesCount > 0 ? Math . min ( Math . pow ( 2 , this . _retriesCount ) * 1000 , 32 * 1000 ) : 0 ; logger . log ( 'info' , 'Restarting RPC WebSocket connection...' , { market : this . _options . marketName , delayMs } ) ; if ( delayMs > 0 ) { await wait ( delayMs ) ; } this . _retriesCount ++ ; this . _wsSubsMeta . clear ( ) ; this . _clearTimers ( ) ; this . _resetCurrentState ( ) ; this . _connectAndStreamData ( ) ; } private _clearTimers ( ) { if ( this . _publishTID !== undefined ) { clearTimeout ( this . _publishTID ) ; } if ( this . _pingTID !== undefined ) { clearInterval ( this . _pingTID ) ; } if ( this . _staleConnectionTID !== undefined ) { clearInterval ( this . _staleConnectionTID ) ; } } private _resetCurrentState ( ) { this . _accountsData = { asks : undefined , bids : undefined , eventQueue : undefined } ; this . _slotStartTimestamp = undefined ; this . _currentSlot = undefined ; this . _state = 'PRISTINE' ; } private _subscribeToAccountsNotifications ( ws ) { for ( const meta of this . _accountsMeta ) { if ( ws . readyState !== ws . OPEN ) { logger . log ( 'warn' , 'Failed to subscribe to accounts notifications' , { market : this . _options . marketName , wsState : ws . readyState } ) ; ws . close ( 1000 , 'Failed to subscribe to accounts notification' ) ; return ; } this . _sendMessage ( ws , { jsonrpc : '2.0' , id : meta . reqId , method : 'accountSubscribe' , params : [ meta . address , { encoding : 'base64' , commitment : this . _options . commitment } ] } ) ; } } private _subscribeToHeartbeat ( ws ) { this . _sendMessage ( ws , { jsonrpc : '2.0' , id : 999 , method : 'slotSubscribe' , params : [ ] } ) ; } private _sendPeriodicPings ( ws ) { if ( this . _pingTID ) { clearInterval ( this . _pingTID ) ; } this . _pingTID = setInterval ( ( ) => { this . _sendMessage ( ws , { jsonrpc : '2.0' , method : 'ping' , params : null } ) ; } , 3 * 1000 ) ; } private _monitorConnectionIfStale ( ws ) { if ( this . _staleConnectionTID ) { clearInterval ( this . _staleConnectionTID ) ; } this . _staleConnectionTID = setInterval ( ( ) => { if ( this . _receivedMessagesCount === 0 ) { logger . log ( 'info' , `Did not received any messages within 120s timeout, terminating connection...` , { market : this . _options . marketName } ) ; ws . terminate ( ) ; } this . _receivedMessagesCount = 0 ; } , 30 * 1000 ) ; } private _sendMessage ( ws , message ) { if ( ws . readyState !== ws . OPEN ) { return ; } ws . send ( JSON . stringify ( message ) , ( err ) => { if ( err !== undefined ) { logger . log ( 'warn' , `WS send error: ${ err . message }` ) ; ws . terminate ( ) ; } } ) ; } private _publish = ( ) => { this . _state = 'PUBLISHED' ; const now = new Date ( ) . valueOf ( ) ; if ( this . _slotStartTimestamp !== undefined ) { const slotTimespan = now - this . _slotStartTimestamp ; if ( slotTimespan > 400 ) { logger . log ( 'debug' , `Slow accounts notification, slot ${ this . _currentSlot }, ${ slotTimespan }ms` , { market : this . _options . marketName } ) ; } } this . _retriesCount = 0 ; this . onAccountsChange ( { accountsData : this . _accountsData , slot : this . _currentSlot ! } ) ; this . _accountsData = { asks : undefined , bids : undefined , eventQueue : undefined } ; this . _slotStartTimestamp = undefined ; if ( this . _publishTID !== undefined ) { clearTimeout ( this . _publishTID ) ; this . _publishTID = undefined ; } } ; private _restartPublishTimer ( ) { if ( this . _publishTID !== undefined ) { clearTimeout ( this . _publishTID ) ; } this . _publishTID = setTimeout ( ( ) => { this . _publish ( ) ; } , 10 * 1000 ) ; } private _receivedDataForAllAccounts ( ) { return ( this . _accountsData . bids !== undefined && this . _accountsData . asks !== undefined && this . _accountsData . eventQueue !== undefined ) ; } private _resetPendingNotificationState ( ) { this . _resetCurrentState ( ) ; if ( this . _publishTID !== undefined ) { clearTimeout ( this . _publishTID ) ; } } private _update ( accountName , accountData , slot ) { if ( logger . level === 'debug' ) { logger . log ( 'debug' , `Received ${ accountName } account update, current state ${ this . _state }` , { market : this . _options . marketName , slot } ) ; } if ( this . _state === 'PUBLISHED' ) { if ( slot < this . _currentSlot ! ) { logger . log ( 'warn' , `Stale notification, current slot ${ this . _currentSlot }, update slot: ${ slot }, ignoring...` , { market : this . _options . marketName } ) ; } else if ( slot > this . _currentSlot ! ) { this . _state = 'PRISTINE' ; } } if ( this . _state === 'PRISTINE' ) { if ( this . _currentSlot === slot ) { logger . log ( 'warn' , 'Ignoring WS account notification' , { market : this . _options . marketName } ) ; return ; } else { this . _currentSlot = slot ; this . _state = 'PENDING' ; this . _slotStartTimestamp = new Date ( ) . valueOf ( ) ; } } if ( this . _state === 'PENDING' ) { this . _restartPublishTimer ( ) ; if ( slot === this . _currentSlot ) { if ( this . _accountsData [ accountName ] !== undefined ) { logger . log ( 'warn' , `Received second update for ${ accountName } account for the same slot` , { slot , market : this . _options . marketName , accountName } ) ; } this . _accountsData [ accountName ] = accountData ; if ( this . _receivedDataForAllAccounts ( ) ) { this . _publish ( ) ; } } else if ( slot > this . _currentSlot ! ) { this . _publish ( ) ; this . _update ( accountName , accountData , slot ) ; } else { logger . log ( 'warn' , `Out of order notification for PENDING event: current slot ${ this . _currentSlot }, update slot: ${ slot }, resetting...` , { market : this . _options . marketName } ) ; this . _resetPendingNotificationState ( ) ; } } } } export type AccountsNotification = AccountsNotificationPayload ; export type AccountsNotificationPayload = { readonly accountsData : AccountsData ; readonly slot : number ; } ; export type AccountName = 'bids' | 'asks' | 'eventQueue' ; export type AccountsData = { [ key in AccountName ] ? : Buffer ; } ;