npm-packages/xrpio
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Added several improvements to failed read/write recovery strategies

Browse Source
This commit is contained in:
nitowa
2023-05-13 01:18:43 +02:00
parent b229e70504
commit 3fea9e1ac7
6 changed files with 78 additions and 216 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/xrpIO/ripple-binding.ts
-197
View File
@@ -1,197 +0,0 @@
import { Memo, Wallet } from '../util/types'
import { MIN_XRP_FEE, MIN_XRP_TX_VALUE } from '../util/protocol.constants';
import { Instructions, LedgerClosedEvent, RippleAPI } from 'ripple-lib'
import { Payment } from 'ripple-lib/dist/npm/transaction/payment';
import * as zlib from 'zlib'
import * as util from 'util'
const chunkString = (str: string, length: number) => str.match(new RegExp('.{1,' + length + '}', 'gs'));
const PAYLOAD_SIZE = 925
const debug = false
const cloneApi = async (api: RippleAPI): Promise<RippleAPI> => {
try {
const subApi = new RippleAPI({ server: api.connection['_url'] })
await subApi.connect()
return subApi
} catch (e) {
if (debug) {
console.log("CLONEAPI ERR", e)
}
return await cloneApi(api)
}
}
export const getLatestSequence = async (api: RippleAPI, accountAddress: string): Promise<number> => {
if (debug) console.log("Getting acc info for", accountAddress)
const accountInfo = await api.getAccountInfo(accountAddress, {})
return Number(accountInfo.sequence - 1)
}
const compressB64 = async (data: string) => (await util.promisify(zlib.deflate)(Buffer.from(data, 'utf-8'))).toString('base64')
const decompressB64 = async (data: string) => (await util.promisify(zlib.inflate)(Buffer.from(data, 'base64'))).toString('utf-8')
const sendReliably = (api: RippleAPI, signed: any, preparedPayment,): Promise<any> => new Promise((res, rej) => {
const ledgerClosedCallback = async (event: LedgerClosedEvent) => {
let status
try {
status = await api.getTransaction(signed.id, {
minLedgerVersion: 25235454
})
} catch (e) {
// Typical error when the tx hasn't been validated yet:
if ((e as Error).name !== 'MissingLedgerHistoryError') {
//console.log(e)
}
if (event.ledger_index > preparedPayment.instructions.maxLedgerVersion + 3) {
// Assumptions:
// - We are still connected to the same rippled server
// - No ledger gaps occurred
// - All ledgers between the time we submitted the tx and now have been checked for the tx
status = {
finalResult: 'Transaction was not, and never will be, included in a validated ledger'
}
return rej(status);
} else {
// Check again later:
api.connection.once('ledgerClosed', ledgerClosedCallback)
return
}
}
return res(status)
}
api.connection.once('ledgerClosed', ledgerClosedCallback)
})
export const sendPayment = async (api: RippleAPI, data: Memo[], from: string, to: string, secret: string, sequence: number) => {
if (debug) console.log("Sending payment with seq", sequence)
const options: Instructions = {
maxLedgerVersionOffset: 5,
fee: MIN_XRP_FEE,
sequence: sequence,
};
const payment: Payment = {
source: {
address: from,
maxAmount: {
value: MIN_XRP_TX_VALUE,
currency: 'XRP'
},
},
destination: {
address: to,
amount: {
value: MIN_XRP_TX_VALUE,
currency: 'XRP'
},
},
memos: data,
};
const _api = await cloneApi(api)
try {
const prepared = await _api.preparePayment(from, payment, options)
const signed = _api.sign(prepared.txJSON, secret);
const txHash = await _api.submit(signed.signedTransaction)
//if(debug) console.log("Transaction submitted", txHash)
await sendReliably(_api, signed, prepared)
return txHash
} catch (error) {
if (debug)
console.log("SENDPAYMENT ERROR", error)
throw error
} finally {
_api.disconnect()
}
}
export const getTransactions = async (api: RippleAPI, address: string, minLedgerVersion: number = 25235454): Promise<any[]> => {
const txs = await api.getTransactions(address, {
minLedgerVersion: minLedgerVersion,
earliestFirst: true,
excludeFailures: true,
})
return txs
}
export const writeRaw = async (api: RippleAPI, data: Memo, from:
string, to: string, secret: string, sequence?: number): Promise<string> => {
//if (memoSize(data) > 1000) throw new Error("data length exceeds capacity")
try {
if (!sequence) {
const accountInfo = await getLatestSequence(api, from)
sequence = accountInfo + 1
}
const resp = await sendPayment(api, [data], from, to, secret, sequence)
return resp['tx_json'].hash
} catch (error) {
if (debug) {
console.log("WRITERAW ERR", error);
}
throw error
}
}
export const readRaw = async (api: RippleAPI, hash: string): Promise<Memo> => {
api = await cloneApi(api)
let tx
try {
tx = await api.getTransaction(hash, {
minLedgerVersion: 25235454
})
} catch (e) {
// if(debug){
console.log("READRAW ERR", e)
api.isConnected
// }
throw e
} finally {
await api.disconnect()
}
if (!tx || !tx.specification || !tx.specification['memos'] || !tx.specification['memos'][0]) {
console.log(tx)
throw new Error('Invalid Transaction ' + hash)
}
return tx.specification['memos'][0]
}
export const subscribe = async (api: RippleAPI, address: string, callback: (tx) => any) => {
api.connection.on('transaction', (tx) => callback(tx))
await api.connection.request({
command: 'subscribe',
accounts: [address],
})
}
export const treeWrite = async (api: RippleAPI, data: string, wallet: Wallet, to: string, format: 'L' | 'N' = 'L'): Promise<string> => {
data = await compressB64(data)
const chunks = chunkString(data, PAYLOAD_SIZE)
const latestSequence = await getLatestSequence(api, wallet.address)
const hashes = await Promise.all(Object.entries(chunks).map(([i, chunk]) => writeRaw(api, { data: chunk, format: format }, wallet.address, to, wallet.secret, latestSequence + Number(i) + 1)))
if (hashes.length === 1) {
return hashes[0]
}
return await treeWrite(api, JSON.stringify(hashes), wallet, to, 'N')
}
export const treeRead = async (api: RippleAPI, hashes: string[]): Promise<string> => {
const memos = await Promise.all(hashes.map(hash => readRaw(api, hash)))
const payload: string = await decompressB64(memos.map(memo => memo.data).join(''))
if (memos.some(memo => memo.format === 'N')) {
return await treeRead(api, JSON.parse(payload))
}
return payload
}
src/xrpIO/xrpl-binding.ts
+31 -11
View File
@@ -1,10 +1,10 @@
import { defaultOptions, Memo, Options } from '../util/types'
import { Client, Payment, RippledError, TxResponse, Wallet } from 'xrpl'
import { Client, DisconnectedError, Payment, TxResponse, Wallet, } from 'xrpl'
import * as zlib from 'zlib'
import * as util from 'util'
import { NON_ZERO_TX_HASH } from '../util/protocol.constants'
import { ERR_BAD_TX_HASH, ERR_NO_VERIFY_OWNER } from '../util/errors'
import { BadTxHashError, CannotVerifyOwnerError } from '../util/errors'
const compressB64 = async (data: string) => (await util.promisify(zlib.deflate)(Buffer.from(data, 'utf-8'))).toString('base64')
const decompressB64 = async (data: string) => (await util.promisify(zlib.inflate)(Buffer.from(data, 'base64'))).toString('utf-8')
@@ -30,6 +30,8 @@ export class xrpIO {
this.options.readMaxRetry = options.readMaxRetry ? Number(options.readMaxRetry) : defaultOptions.readMaxRetry
this.options.readRetryTimeout = options.readRetryTimeout ? Number(options.readRetryTimeout) : defaultOptions.readRetryTimeout
this.options.readFreshApi = options.readFreshApi ? Boolean(options.readFreshApi) : defaultOptions.readFreshApi
this.options.writeMaxRetry = options.writeMaxRetry ? Number(options.readFreshApi) : defaultOptions.writeMaxRetry
this.options.writeRetryTimeout = options.writeRetryTimeout ? Number(options.writeRetryTimeout) : defaultOptions.writeRetryTimeout
this.api = new Client(server, {
connectionTimeout: this.options.connectionTimeout
@@ -68,7 +70,7 @@ export class xrpIO {
}
}
public async sendPayment(data: Memo, to: string, secret: string, sequence?: number, amount: string = "1"): Promise<TxResponse> {
public async sendPayment(data: Memo, to: string, secret: string, sequence?: number, amount: string = "1", retry = 0): Promise<TxResponse> {
const wallet = Wallet.fromSecret(secret)
this.dbg("Sending payment", wallet.address, '->', to)
@@ -88,17 +90,35 @@ export class xrpIO {
}]
})
let response: TxResponse;
try {
const response = await _api.submitAndWait(payment, { wallet })
this.dbg("Tx finalized", response.result.hash, response.result.Sequence)
return response
response = await _api.submitAndWait(payment, { wallet })
} catch (error: any) {
this.dbg("SENDPAYMENT ERROR", error)
console.log("SENDPAYMENT ERROR", error)
if(error instanceof DisconnectedError){
//Usually caused by hitting rate limits. Recoverable.
if(this.options.writeMaxRetry != -1 && retry >= this.options.writeMaxRetry){
//exceeded retry quota
throw error
}
await new Promise(res => setTimeout(res, this.options.writeRetryTimeout))
return await this.sendPayment(data, to, secret, sequence, amount, retry+1)
}
throw error
}finally{
await _api.disconnect()
}
//We cannot afford this payment. Irrecoverable error
if(response.result.meta && response.result.meta['TransactionResult'] && response.result.meta['TransactionResult'] === 'tecUNFUNDED_PAYMENT'){
throw new Error(`Insufficient funds to send transaction. See tx ${response.result.hash}`)
}
this.dbg("Tx finalized", response.result.hash, response.result.Sequence)
return response
}
public async writeRaw(data: Memo, to: string, secret: string, sequence?: number, amount: string = "1"): Promise<string> {
@@ -109,7 +129,7 @@ export class xrpIO {
public async getTransaction(hash: string, retry = 0): Promise<TxResponse> {
if (!NON_ZERO_TX_HASH.test(hash)) {
throw ERR_BAD_TX_HASH(hash)
throw new BadTxHashError(hash)
}
this.dbg("Getting Tx", hash)
@@ -163,12 +183,12 @@ export class xrpIO {
public async readRaw(hash: string, verifyOwner?: string): Promise<Memo> {
if (!NON_ZERO_TX_HASH.test(hash)) {
throw ERR_BAD_TX_HASH(hash)
throw new BadTxHashError(hash)
}
const tx = await this.getTransaction(hash)
if(verifyOwner && tx.result.Account != verifyOwner){
throw ERR_NO_VERIFY_OWNER(hash, tx.result.Account, verifyOwner)
throw new CannotVerifyOwnerError(hash, tx.result.Account, verifyOwner)
}
const memo = tx.result.Memos[0].Memo
@@ -198,7 +218,7 @@ export class xrpIO {
public async treeRead(hashes: string[], verifyOwner?:string): Promise<string> {
const bad_hash = hashes.find(hash => !NON_ZERO_TX_HASH.test(hash))
if (bad_hash)
throw ERR_BAD_TX_HASH(bad_hash)
throw new BadTxHashError(bad_hash)
const memos = await Promise.all(hashes.map(hash => this.readRaw(hash, verifyOwner)))
const payload: string = await decompressB64(memos.map(memo => memo.data).join(''))