8184453898

Committed 07 Mar 2024 07:27AM UTC coverage: 78.668% (-0.04%) from 78.703%

Build # 8184453898

Build Type

push

github

Committed by

vgrichina

Commit Message

Avoid fetching block stream twice

Run Details

392 of 591 branches covered (66.33%)

Branch coverage included in aggregate %.

0 of 1 new or added line in 1 file covered. (0.0%)

1 existing line in 1 file now uncovered.

1404 of 1692 relevant lines covered (82.98%)

15710.91 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

11.54

/utils/submit-transaction.js

const { redisBlockStream } = require('./redis-block-stream');
const { SignedTransaction, BORSH_SCHEMA } = require('../data-model');
const { deserialize } = require('borsh');
const LRU = require('lru-cache');
const EventEmitter = require('events');
const debug = require('debug')('submit-transaction');

// TODO: Dedupe with json-rpc.js
const NODE_URL = process.env.FAST_NEAR_NODE_URL || 'https://rpc.mainnet.near.org';

async function sendJsonRpc(method, params) {
    const res  = await fetch(`${NODE_URL}`, {
        method: 'POST',
        headers: {
            'Content-Type': 'application/json'
        },
        body: JSON.stringify({
            jsonrpc: '2.0',
            id: 'fast-near',
            method,
            params
        })
    });
    const json = await res.json();
    if (json.error) {
        // TODO: Special RPCError?
        const error = new Error(`RPC Error: ${json.error.message}: ${json.error.data})`);
        error.jsonRpcError = json.error;
        throw error;
    }
    return json.result;
}

async function submitTransactionAsync(transactionData) {
    return await sendJsonRpc('broadcast_tx_async', [transactionData.toString('base64')]);
}

// TODO: Use our own DB instead
async function txStatus(txHash, accountId) {
    return await sendJsonRpc('tx', [txHash, accountId]);
}

let blockStream;
const txEventEmitter = new EventEmitter();
const txCache = new LRU({ max: 10000, maxAge: 1000 * 60 * 60 });

async function submitTransaction(transactionData) {
    if (!blockStream) {
        blockStream = true; // NOTE: Prevent reentrancy
        const startBlockHeight = (await (await fetch(`${NODE_URL}/status`, {
            method: 'GET',
            headers: {
                'Content-Type': 'application/json'
            }
        })).json()).sync_info.latest_block_height;
        debug('startBlockHeight:', startBlockHeight);

        const redisUrl = process.env.BLOCKS_REDIS_URL;
        debug('redisUrl:', redisUrl);
        blockStream = redisBlockStream({ startBlockHeight, redisUrl, batchSize: 1 });
        await new Promise(async (resolve, reject) => {
            (async () => {
                let streamStarted = false;
                for await (const { block, shards } of blockStream) {
                    if (!streamStarted) {
                        streamStarted = true;
                        // NOTE: We are waiting for block stream to start and then it continues
                        resolve();
                    }

                    for (const { chunk } of shards) {
                        if (!chunk) {
                            continue;
                        }

                        for await (const { transaction, outcome } of chunk.transactions) {
                            const txHash = transaction.hash;
                            txCache.set(txHash, { transaction, outcome });
                        }
                    }
                    debug('Block:', block.header.height);
                    txEventEmitter.emit('block');
                }
            })().catch(reject);
        });
    }

    const txHash = await submitTransactionAsync(transactionData);
    debug('Transaction posted:', txHash);
    if (txCache.has(txHash)) {
        return txCache.get(txHash);
    }

    const SUBMIT_TX_STATUS_CHECK_TIMEOUT = 1000 * 10;
    const SUBMIT_TOTAL_TIMEOUT = 1000 * 45;
    const { transaction, outcome } = await new Promise((resolve, reject) => {
        let blockCallback;
        const subscribe = () => txEventEmitter.on('block', blockCallback);
        const unsubscribe = () => blockCallback && (txEventEmitter.off('block', blockCallback), blockCallback = null);
        setTimeout(() => (unsubscribe(), reject(new Error(`Taking more than ${SUBMIT_TOTAL_TIMEOUT}ms to submit transaction`))), SUBMIT_TOTAL_TIMEOUT);

        // NOTE: This is necessary in case transaction already landed before but we don't know
        setTimeout(async function checkStatus() {
            if (!blockCallback) return;

            const { transaction: { signerId } } = deserialize(BORSH_SCHEMA, SignedTransaction, transactionData);
            debug('Checking txStatus', txHash, signerId);
            try {
                const result = await txStatus(txHash, signerId);
                const { transaction, transaction_outcome } = result;
                debug('txStatus result:', result);

                if (blockCallback) {
                    unsubscribe();
                    resolve({ transaction, outcome: transaction_outcome });
                }
            } catch (error) {
                debug('txStatus error:', error);
                if (error.jsonRpcError) {
                    const { name, cause } = error.jsonRpcError;
                    if (name == 'HANDLER_ERROR' && cause.name ==  'UNKNOWN_TRANSACTION') {
                        setTimeout(checkStatus, SUBMIT_TX_STATUS_CHECK_TIMEOUT);
                        return;
                    }
                    unsubscribe();
                    reject(error);
                }
            }
        }, SUBMIT_TX_STATUS_CHECK_TIMEOUT);

        blockCallback = () => {
            if (txCache.has(txHash)) {
                unsubscribe();
                resolve(txCache.get(txHash));
            }
        };
        subscribe();
    });
    
    const result = {
        // TODO: Other fields
        transaction,
        transaction_outcome: outcome
    }
    console.log('result:', result);
    return result;
}

module.exports = { submitTransaction };

1	const { redisBlockStream } = require('./redis-block-stream');	5✔
2	const { SignedTransaction, BORSH_SCHEMA } = require('../data-model');	5✔
3	const { deserialize } = require('borsh');	5✔
4	const LRU = require('lru-cache');	5✔
5	const EventEmitter = require('events');	5✔
6	const debug = require('debug')('submit-transaction');	5✔
7
8	// TODO: Dedupe with json-rpc.js
9	const NODE_URL = process.env.FAST_NEAR_NODE_URL \|\| 'https://rpc.mainnet.near.org';	5✔
10
11	async function sendJsonRpc(method, params) {
12	const res = await fetch(`${NODE_URL}`, {	×
13	method: 'POST',
14	headers: {
15	'Content-Type': 'application/json'
16	},
17	body: JSON.stringify({
18	jsonrpc: '2.0',
19	id: 'fast-near',
20	method,
21	params
22	})
23	});
24	const json = await res.json();	×
25	if (json.error) {	×
26	// TODO: Special RPCError?
27	const error = new Error(`RPC Error: ${json.error.message}: ${json.error.data})`);	×
28	error.jsonRpcError = json.error;	×
29	throw error;	×
30	}
31	return json.result;	×
32	}
33
34	async function submitTransactionAsync(transactionData) {
35	return await sendJsonRpc('broadcast_tx_async', [transactionData.toString('base64')]);	×
36	}
37
38	// TODO: Use our own DB instead
39	async function txStatus(txHash, accountId) {
40	return await sendJsonRpc('tx', [txHash, accountId]);	×
41	}
42
43	let blockStream;
44	const txEventEmitter = new EventEmitter();	5✔
45	const txCache = new LRU({ max: 10000, maxAge: 1000 * 60 * 60 });	5✔
46
47	async function submitTransaction(transactionData) {
48	if (!blockStream) {	×
NEW 49	blockStream = true; // NOTE: Prevent reentrancy	×
UNCOV 50	const startBlockHeight = (await (await fetch(`${NODE_URL}/status`, {	×
51	method: 'GET',
52	headers: {
53	'Content-Type': 'application/json'
54	}
55	})).json()).sync_info.latest_block_height;
56	debug('startBlockHeight:', startBlockHeight);	×
57
58	const redisUrl = process.env.BLOCKS_REDIS_URL;	×
59	debug('redisUrl:', redisUrl);	×
60	blockStream = redisBlockStream({ startBlockHeight, redisUrl, batchSize: 1 });	×
61	await new Promise(async (resolve, reject) => {	×
62	(async () => {	×
63	let streamStarted = false;	×
64	for await (const { block, shards } of blockStream) {	×
65	if (!streamStarted) {	×
66	streamStarted = true;	×
67	// NOTE: We are waiting for block stream to start and then it continues
68	resolve();	×
69	}
70
71	for (const { chunk } of shards) {	×
72	if (!chunk) {	×
73	continue;	×
74	}
75
76	for await (const { transaction, outcome } of chunk.transactions) {	×
77	const txHash = transaction.hash;	×
78	txCache.set(txHash, { transaction, outcome });	×
79	}
80	}
81	debug('Block:', block.header.height);	×
82	txEventEmitter.emit('block');	×
83	}
84	})().catch(reject);
85	});
86	}
87
88	const txHash = await submitTransactionAsync(transactionData);	×
89	debug('Transaction posted:', txHash);	×
90	if (txCache.has(txHash)) {	×
91	return txCache.get(txHash);	×
92	}
93
94	const SUBMIT_TX_STATUS_CHECK_TIMEOUT = 1000 * 10;	×
95	const SUBMIT_TOTAL_TIMEOUT = 1000 * 45;	×
96	const { transaction, outcome } = await new Promise((resolve, reject) => {	×
97	let blockCallback;
98	const subscribe = () => txEventEmitter.on('block', blockCallback);	×
99	const unsubscribe = () => blockCallback && (txEventEmitter.off('block', blockCallback), blockCallback = null);	×
100	setTimeout(() => (unsubscribe(), reject(new Error(`Taking more than ${SUBMIT_TOTAL_TIMEOUT}ms to submit transaction`))), SUBMIT_TOTAL_TIMEOUT);	×
101
102	// NOTE: This is necessary in case transaction already landed before but we don't know
103	setTimeout(async function checkStatus() {	×
104	if (!blockCallback) return;	×
105
106	const { transaction: { signerId } } = deserialize(BORSH_SCHEMA, SignedTransaction, transactionData);	×
107	debug('Checking txStatus', txHash, signerId);	×
108	try {	×
109	const result = await txStatus(txHash, signerId);	×
110	const { transaction, transaction_outcome } = result;	×
111	debug('txStatus result:', result);	×
112
113	if (blockCallback) {	×
114	unsubscribe();	×
115	resolve({ transaction, outcome: transaction_outcome });	×
116	}
117	} catch (error) {
118	debug('txStatus error:', error);	×
119	if (error.jsonRpcError) {	×
120	const { name, cause } = error.jsonRpcError;	×
121	if (name == 'HANDLER_ERROR' && cause.name == 'UNKNOWN_TRANSACTION') {	×
122	setTimeout(checkStatus, SUBMIT_TX_STATUS_CHECK_TIMEOUT);	×
123	return;	×
124	}
125	unsubscribe();	×
126	reject(error);	×
127	}
128	}
129	}, SUBMIT_TX_STATUS_CHECK_TIMEOUT);
130
131	blockCallback = () => {	×
132	if (txCache.has(txHash)) {	×
133	unsubscribe();	×
134	resolve(txCache.get(txHash));	×
135	}
136	};
137	subscribe();	×
138	});
139
140	const result = {	×
141	// TODO: Other fields
142	transaction,
143	transaction_outcome: outcome
144	}
145	console.log('result:', result);	×
146	return result;	×
147	}
148
149	module.exports = { submitTransaction };	5✔

vgrichina / fast-near / 8184453898

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous