4701730719

Build Type

push

github

Committed by GitHub

Commit Message

fix: warning logger level for RPC proxy errors (#1612)

Run Details

2085 of 3213 branches covered (64.89%)

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

7314 of 9547 relevant lines covered (76.61%)

1569.91 hits per line

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

76.16

/src/api/routes/core-node-rpc-proxy.ts

import * as express from 'express';
import * as cors from 'cors';
import { createProxyMiddleware, Options, responseInterceptor } from 'http-proxy-middleware';
import { logError, logger, parsePort, pipelineAsync, REPO_DIR } from '../../helpers';
import { Agent } from 'http';
import * as fs from 'fs';
import * as path from 'path';
import { asyncHandler } from '../async-handler';
import * as chokidar from 'chokidar';
import * as jsoncParser from 'jsonc-parser';
import fetch, { RequestInit } from 'node-fetch';
import { PgStore } from '../../datastore/pg-store';

function GetStacksNodeProxyEndpoint() {
  // Use STACKS_CORE_PROXY env vars if available, otherwise fallback to `STACKS_CORE_RPC
  const proxyHost =
    process.env['STACKS_CORE_PROXY_HOST'] ?? process.env['STACKS_CORE_RPC_HOST'] ?? '';
  const proxyPort =
    parsePort(process.env['STACKS_CORE_PROXY_PORT'] ?? process.env['STACKS_CORE_RPC_PORT']) ?? 0;
  return `${proxyHost}:${proxyPort}`;
}

export function createCoreNodeRpcProxyRouter(db: PgStore): express.Router {
  const router = express.Router();
  router.use(cors());

  const stacksNodeRpcEndpoint = GetStacksNodeProxyEndpoint();

  logger.info(`/v2/* proxying to: ${stacksNodeRpcEndpoint}`);

  // Note: while keep-alive may result in some performance improvements with the stacks-node http server,
  // it can also cause request distribution issues when proxying to a pool of stacks-nodes. See:
  // https://github.com/hirosystems/stacks-blockchain-api/issues/756
  const httpAgent = new Agent({
    // keepAlive: true,
    keepAlive: false, // `false` is the default -- set it explicitly for readability anyway.
    // keepAliveMsecs: 60000,
    maxSockets: 200,
    maxTotalSockets: 400,
  });

  const PROXY_CACHE_CONTROL_FILE_ENV_VAR = 'STACKS_API_PROXY_CACHE_CONTROL_FILE';
  let proxyCacheControlFile = '.proxy-cache-control.json';
  if (process.env[PROXY_CACHE_CONTROL_FILE_ENV_VAR]) {
    proxyCacheControlFile = process.env[PROXY_CACHE_CONTROL_FILE_ENV_VAR] as string;
    logger.info(`Using ${proxyCacheControlFile}`);
  }
  const cacheControlFileWatcher = chokidar.watch(proxyCacheControlFile, {
    persistent: false,
    useFsEvents: false,
    ignoreInitial: true,
  });
  let pathCacheOptions = new Map<RegExp, string | null>();

  const updatePathCacheOptions = () => {
    try {
      const configContent: { paths: Record<string, string> } = jsoncParser.parse(
        fs.readFileSync(proxyCacheControlFile, 'utf8')
      );
      pathCacheOptions = new Map(
        Object.entries(configContent.paths).map(([k, v]) => [RegExp(k), v])
      );
    } catch (error) {
      pathCacheOptions.clear();
      logger.error(`Error reading changes from ${proxyCacheControlFile}`, error);
    }
  };
  updatePathCacheOptions();
  cacheControlFileWatcher.on('all', (eventName, path, stats) => {
    updatePathCacheOptions();
  });

  const getCacheControlHeader = (statusCode: number, url: string): string | null => {
    if (statusCode < 200 || statusCode > 299) {
      return null;
    }
    for (const [regexp, cacheControl] of pathCacheOptions.entries()) {
      if (cacheControl && regexp.test(url)) {
        return cacheControl;
      }
    }
    return null;
  };

  /**
   * Check for any extra endpoints that have been configured for performing a "multicast" for a tx submission.
   */
  async function getExtraTxPostEndpoints(): Promise<string[] | false> {
    const STACKS_API_EXTRA_TX_ENDPOINTS_FILE_ENV_VAR = 'STACKS_API_EXTRA_TX_ENDPOINTS_FILE';
    const extraEndpointsEnvVar = process.env[STACKS_API_EXTRA_TX_ENDPOINTS_FILE_ENV_VAR];
    if (!extraEndpointsEnvVar) {
      return false;
    }
    const filePath = path.resolve(REPO_DIR, extraEndpointsEnvVar);
    let fileContents: string;
    try {
      fileContents = await fs.promises.readFile(filePath, { encoding: 'utf8' });
    } catch (error) {
      logError(`Error reading ${STACKS_API_EXTRA_TX_ENDPOINTS_FILE_ENV_VAR}: ${error}`, error);
      return false;
    }
    const endpoints = fileContents
      .split(/\r?\n/)
      .map(r => r.trim())
      .filter(r => !r.startsWith('#') && r.length !== 0);
    if (endpoints.length === 0) {
      return false;
    }
    return endpoints;
  }

  /**
   * Reads an http request stream into a Buffer.
   */
  async function readRequestBody(req: express.Request, maxSizeBytes = Infinity): Promise<Buffer> {
    return new Promise((resolve, reject) => {
      let resultBuffer: Buffer = Buffer.alloc(0);
      req.on('data', chunk => {
        if (!Buffer.isBuffer(chunk)) {
          reject(
            new Error(
              `Expected request body chunks to be Buffer, received ${chunk.constructor.name}`
            )
          );
          req.destroy();
          return;
        }
        resultBuffer = resultBuffer.length === 0 ? chunk : Buffer.concat([resultBuffer, chunk]);
        if (resultBuffer.byteLength >= maxSizeBytes) {
          reject(new Error(`Request body exceeded max byte size`));
          req.destroy();
          return;
        }
      });
      req.on('end', () => {
        if (!req.complete) {
          return reject(
            new Error('The connection was terminated while the message was still being sent')
          );
        }
        resolve(resultBuffer);
      });
      req.on('error', error => reject(error));
    });
  }

  /**
   * Logs a transaction broadcast event alongside the current block height.
   */
  async function logTxBroadcast(response: string): Promise<void> {
    try {
      const blockHeightQuery = await db.getCurrentBlockHeight();
      if (!blockHeightQuery.found) {
        return;
      }
      const blockHeight = blockHeightQuery.result;
      // Strip wrapping double quotes (if any)
      const txId = response.replace(/^"(.*)"$/, '$1');
      logger.info('Transaction broadcasted', {
        txid: `0x${txId}`,
        first_broadcast_at_stacks_height: blockHeight,
      });
    } catch (error) {
      logError(`Error logging tx broadcast: ${error}`, error);
    }
  }

  router.post(
    '/transactions',
    asyncHandler(async (req, res, next) => {
      const extraEndpoints = await getExtraTxPostEndpoints();
      if (!extraEndpoints) {
        next();
        return;
      }
      const endpoints = [
        // The primary proxy endpoint (the http response from this one will be returned to the client)
        `http://${stacksNodeRpcEndpoint}/v2/transactions`,
      ];
      endpoints.push(...extraEndpoints);
      logger.info(`Overriding POST /v2/transactions to multicast to ${endpoints.join(',')}}`);
      const maxBodySize = 10_000_000; // 10 MB max POST body size
      const reqBody = await readRequestBody(req, maxBodySize);
      const reqHeaders: string[][] = [];
      for (let i = 0; i < req.rawHeaders.length; i += 2) {
        reqHeaders.push([req.rawHeaders[i], req.rawHeaders[i + 1]]);
      }
      const postFn = async (endpoint: string) => {
        const reqOpts: RequestInit = {
          method: 'POST',
          agent: httpAgent,
          body: reqBody,
          headers: reqHeaders,
        };
        const proxyResult = await fetch(endpoint, reqOpts);
        return proxyResult;
      };

      // Here's were we "multicast" the `/v2/transaction` POST, by concurrently sending the http request to all configured endpoints.
      const results = await Promise.allSettled(endpoints.map(endpoint => postFn(endpoint)));

      // Only the first (non-extra) endpoint http response is proxied back through to the client, so ensure any errors from requests
      // to the extra endpoints are logged.
      results.slice(1).forEach(p => {
        if (p.status === 'rejected') {
          logError(`Error during POST /v2/transaction to extra endpoint: ${p.reason}`, p.reason);
        } else {
          if (!p.value.ok) {
            logger.warn(
              `Response ${p.value.status} during POST /v2/transaction to extra endpoint ${p.value.url}`
            );
          }
        }
      });

      // Proxy the result of the (non-extra) http response back to the client.
      const mainResult = results[0];
      if (mainResult.status === 'rejected') {
        logError(
          `Error in primary POST /v2/transaction proxy: ${mainResult.reason}`,
          mainResult.reason
        );
        res.status(500).json({ error: mainResult.reason });
      } else {
        const proxyResp = mainResult.value;
        res.status(proxyResp.status);
        proxyResp.headers.forEach((value, name) => {
          res.setHeader(name, value);
        });
        if (proxyResp.status === 200) {
          // Log the transaction id broadcast, but clone the `Response` first before parsing its body
          // so we don't mess up the original response's `ReadableStream` pointers.
          const parsedTxId: string = await proxyResp.clone().text();
          await logTxBroadcast(parsedTxId);
        }
        await pipelineAsync(proxyResp.body, res);
      }
    })
  );

  const proxyOptions: Options = {
    agent: httpAgent,
    target: `http://${stacksNodeRpcEndpoint}`,
    changeOrigin: true,
    selfHandleResponse: true,
    onProxyRes: responseInterceptor(async (responseBuffer, proxyRes, req, res) => {
      if (req.url !== undefined) {
        const header = getCacheControlHeader(res.statusCode, req.url);
        if (header) {
          res.setHeader('Cache-Control', header);
        }
        const url = new URL(req.url, `http://${req.headers.host}`);
        if (url.pathname === '/v2/transactions' && res.statusCode === 200) {
          await logTxBroadcast(responseBuffer.toString());
        }
      }
      return responseBuffer;
    }),
    onError: (error, req, res) => {
      const msg =
        (error as any).code === 'ECONNREFUSED'
          ? 'core node unresponsive'
          : 'cannot connect to core node';
      res
        .writeHead(502, { 'Content-Type': 'application/json' })
        .end(JSON.stringify({ message: msg, error: error }));
    },
  };

  const stacksNodeRpcProxy = createProxyMiddleware(proxyOptions);

  router.use(stacksNodeRpcProxy);

  return router;
}

1	import * as express from 'express';	32✔
2	import * as cors from 'cors';	32✔
3	import { createProxyMiddleware, Options, responseInterceptor } from 'http-proxy-middleware';	32✔
4	import { logError, logger, parsePort, pipelineAsync, REPO_DIR } from '../../helpers';	32✔
5	import { Agent } from 'http';	32✔
6	import * as fs from 'fs';	32✔
7	import * as path from 'path';	32✔
8	import { asyncHandler } from '../async-handler';	32✔
9	import * as chokidar from 'chokidar';	32✔
10	import * as jsoncParser from 'jsonc-parser';	32✔
11	import fetch, { RequestInit } from 'node-fetch';	32✔
12	import { PgStore } from '../../datastore/pg-store';
13
14	function GetStacksNodeProxyEndpoint() {
15	// Use STACKS_CORE_PROXY env vars if available, otherwise fallback to `STACKS_CORE_RPC
16	const proxyHost =
17	process.env['STACKS_CORE_PROXY_HOST'] ?? process.env['STACKS_CORE_RPC_HOST'] ?? '';	207!
18	const proxyPort =
19	parsePort(process.env['STACKS_CORE_PROXY_PORT'] ?? process.env['STACKS_CORE_RPC_PORT']) ?? 0;	207!
20	return `${proxyHost}:${proxyPort}`;	207✔
21	}
22
23	export function createCoreNodeRpcProxyRouter(db: PgStore): express.Router {	32✔
24	const router = express.Router();	207✔
25	router.use(cors());	207✔
26
27	const stacksNodeRpcEndpoint = GetStacksNodeProxyEndpoint();	207✔
28
29	logger.info(`/v2/* proxying to: ${stacksNodeRpcEndpoint}`);	207✔
30
31	// Note: while keep-alive may result in some performance improvements with the stacks-node http server,
32	// it can also cause request distribution issues when proxying to a pool of stacks-nodes. See:
33	// https://github.com/hirosystems/stacks-blockchain-api/issues/756
34	const httpAgent = new Agent({	207✔
35	// keepAlive: true,
36	keepAlive: false, // `false` is the default -- set it explicitly for readability anyway.
37	// keepAliveMsecs: 60000,
38	maxSockets: 200,
39	maxTotalSockets: 400,
40	});
41
42	const PROXY_CACHE_CONTROL_FILE_ENV_VAR = 'STACKS_API_PROXY_CACHE_CONTROL_FILE';	207✔
43	let proxyCacheControlFile = '.proxy-cache-control.json';	207✔
44	if (process.env[PROXY_CACHE_CONTROL_FILE_ENV_VAR]) {	207!
45	proxyCacheControlFile = process.env[PROXY_CACHE_CONTROL_FILE_ENV_VAR] as string;	×
46	logger.info(`Using ${proxyCacheControlFile}`);	×
47	}
48	const cacheControlFileWatcher = chokidar.watch(proxyCacheControlFile, {	207✔
49	persistent: false,
50	useFsEvents: false,
51	ignoreInitial: true,
52	});
53	let pathCacheOptions = new Map<RegExp, string \| null>();	207✔
54
55	const updatePathCacheOptions = () => {	207✔
56	try {	207✔
57	const configContent: { paths: Record<string, string> } = jsoncParser.parse(	207✔
58	fs.readFileSync(proxyCacheControlFile, 'utf8')
59	);
60	pathCacheOptions = new Map(	207✔
61	Object.entries(configContent.paths).map(([k, v]) => [RegExp(k), v])	1,035✔
62	);
63	} catch (error) {
64	pathCacheOptions.clear();	×
65	logger.error(`Error reading changes from ${proxyCacheControlFile}`, error);	×
66	}
67	};
68	updatePathCacheOptions();	207✔
69	cacheControlFileWatcher.on('all', (eventName, path, stats) => {	207✔
70	updatePathCacheOptions();	×
71	});
72
73	const getCacheControlHeader = (statusCode: number, url: string): string \| null => {	207✔
74	if (statusCode < 200 \|\| statusCode > 299) {	187✔
75	return null;	88✔
76	}
77	for (const [regexp, cacheControl] of pathCacheOptions.entries()) {	99✔
78	if (cacheControl && regexp.test(url)) {	495!
79	return cacheControl;	×
80	}
81	}
82	return null;	99✔
83	};
84
85	/**
86	* Check for any extra endpoints that have been configured for performing a "multicast" for a tx submission.
87	*/
88	async function getExtraTxPostEndpoints(): Promise<string[] \| false> {
89	const STACKS_API_EXTRA_TX_ENDPOINTS_FILE_ENV_VAR = 'STACKS_API_EXTRA_TX_ENDPOINTS_FILE';	34✔
90	const extraEndpointsEnvVar = process.env[STACKS_API_EXTRA_TX_ENDPOINTS_FILE_ENV_VAR];	34✔
91	if (!extraEndpointsEnvVar) {	34✔
92	return false;	33✔
93	}
94	const filePath = path.resolve(REPO_DIR, extraEndpointsEnvVar);	1✔
95	let fileContents: string;
96	try {	1✔
97	fileContents = await fs.promises.readFile(filePath, { encoding: 'utf8' });	1✔
98	} catch (error) {
99	logError(`Error reading ${STACKS_API_EXTRA_TX_ENDPOINTS_FILE_ENV_VAR}: ${error}`, error);	×
100	return false;	×
101	}
102	const endpoints = fileContents	1✔
103	.split(/\r?\n/)
104	.map(r => r.trim())	1✔
105	.filter(r => !r.startsWith('#') && r.length !== 0);	1✔
106	if (endpoints.length === 0) {	1!
107	return false;	×
108	}
109	return endpoints;	1✔
110	}
111
112	/**
113	* Reads an http request stream into a Buffer.
114	*/
115	async function readRequestBody(req: express.Request, maxSizeBytes = Infinity): Promise<Buffer> {	×
116	return new Promise((resolve, reject) => {	1✔
117	let resultBuffer: Buffer = Buffer.alloc(0);	1✔
118	req.on('data', chunk => {	1✔
119	if (!Buffer.isBuffer(chunk)) {	1!
120	reject(	×
121	new Error(
122	`Expected request body chunks to be Buffer, received ${chunk.constructor.name}`
123	)
124	);
125	req.destroy();	×
126	return;	×
127	}
128	resultBuffer = resultBuffer.length === 0 ? chunk : Buffer.concat([resultBuffer, chunk]);	1!
129	if (resultBuffer.byteLength >= maxSizeBytes) {	1!
130	reject(new Error(`Request body exceeded max byte size`));	×
131	req.destroy();	×
132	return;	×
133	}
134	});
135	req.on('end', () => {	1✔
136	if (!req.complete) {	1!
137	return reject(	×
138	new Error('The connection was terminated while the message was still being sent')
139	);
140	}
141	resolve(resultBuffer);	1✔
142	});
143	req.on('error', error => reject(error));	1✔
144	});
145	}
146
147	/**
148	* Logs a transaction broadcast event alongside the current block height.
149	*/
150	async function logTxBroadcast(response: string): Promise<void> {
151	try {	34✔
152	const blockHeightQuery = await db.getCurrentBlockHeight();	34✔
153	if (!blockHeightQuery.found) {	34!
154	return;	×
155	}
156	const blockHeight = blockHeightQuery.result;	34✔
157	// Strip wrapping double quotes (if any)
158	const txId = response.replace(/^"(.*)"$/, '$1');	34✔
159	logger.info('Transaction broadcasted', {	34✔
160	txid: `0x${txId}`,
161	first_broadcast_at_stacks_height: blockHeight,
162	});
163	} catch (error) {
164	logError(`Error logging tx broadcast: ${error}`, error);	×
165	}
166	}
167
168	router.post(	207✔
169	'/transactions',
170	asyncHandler(async (req, res, next) => {
171	const extraEndpoints = await getExtraTxPostEndpoints();	34✔
172	if (!extraEndpoints) {	34✔
173	next();	33✔
174	return;	33✔
175	}
176	const endpoints = [	1✔
177	// The primary proxy endpoint (the http response from this one will be returned to the client)
178	`http://${stacksNodeRpcEndpoint}/v2/transactions`,
179	];
180	endpoints.push(...extraEndpoints);	1✔
181	logger.info(`Overriding POST /v2/transactions to multicast to ${endpoints.join(',')}}`);	1✔
182	const maxBodySize = 10_000_000; // 10 MB max POST body size	1✔
183	const reqBody = await readRequestBody(req, maxBodySize);	1✔
184	const reqHeaders: string[][] = [];	1✔
185	for (let i = 0; i < req.rawHeaders.length; i += 2) {	1✔
186	reqHeaders.push([req.rawHeaders[i], req.rawHeaders[i + 1]]);	6✔
187	}
188	const postFn = async (endpoint: string) => {	1✔
189	const reqOpts: RequestInit = {	2✔
190	method: 'POST',
191	agent: httpAgent,
192	body: reqBody,
193	headers: reqHeaders,
194	};
195	const proxyResult = await fetch(endpoint, reqOpts);	2✔
196	return proxyResult;	2✔
197	};
198
199	// Here's were we "multicast" the `/v2/transaction` POST, by concurrently sending the http request to all configured endpoints.
200	const results = await Promise.allSettled(endpoints.map(endpoint => postFn(endpoint)));	2✔
201
202	// Only the first (non-extra) endpoint http response is proxied back through to the client, so ensure any errors from requests
203	// to the extra endpoints are logged.
204	results.slice(1).forEach(p => {	1✔
205	if (p.status === 'rejected') {	1!
206	logError(`Error during POST /v2/transaction to extra endpoint: ${p.reason}`, p.reason);	×
207	} else {
208	if (!p.value.ok) {	1!
NEW 209	logger.warn(	×
210	`Response ${p.value.status} during POST /v2/transaction to extra endpoint ${p.value.url}`
211	);
212	}
213	}
214	});
215
216	// Proxy the result of the (non-extra) http response back to the client.
217	const mainResult = results[0];	1✔
218	if (mainResult.status === 'rejected') {	1!
219	logError(	×
220	`Error in primary POST /v2/transaction proxy: ${mainResult.reason}`,
221	mainResult.reason
222	);
223	res.status(500).json({ error: mainResult.reason });	×
224	} else {
225	const proxyResp = mainResult.value;	1✔
226	res.status(proxyResp.status);	1✔
227	proxyResp.headers.forEach((value, name) => {	1✔
228	res.setHeader(name, value);	×
229	});
230	if (proxyResp.status === 200) {	1✔
231	// Log the transaction id broadcast, but clone the `Response` first before parsing its body
232	// so we don't mess up the original response's `ReadableStream` pointers.
233	const parsedTxId: string = await proxyResp.clone().text();	1✔
234	await logTxBroadcast(parsedTxId);	1✔
235	}
236	await pipelineAsync(proxyResp.body, res);	1✔
237	}
238	})
239	);
240
241	const proxyOptions: Options = {	207✔
242	agent: httpAgent,
243	target: `http://${stacksNodeRpcEndpoint}`,
244	changeOrigin: true,
245	selfHandleResponse: true,
246	onProxyRes: responseInterceptor(async (responseBuffer, proxyRes, req, res) => {
247	if (req.url !== undefined) {	187✔
248	const header = getCacheControlHeader(res.statusCode, req.url);	187✔
249	if (header) {	187!
250	res.setHeader('Cache-Control', header);	×
251	}
252	const url = new URL(req.url, `http://${req.headers.host}`);	187✔
253	if (url.pathname === '/v2/transactions' && res.statusCode === 200) {	187✔
254	await logTxBroadcast(responseBuffer.toString());	33✔
255	}
256	}
257	return responseBuffer;	187✔
258	}),
259	onError: (error, req, res) => {
260	const msg =
261	(error as any).code === 'ECONNREFUSED'	31!
262	? 'core node unresponsive'
263	: 'cannot connect to core node';
264	res	31✔
265	.writeHead(502, { 'Content-Type': 'application/json' })
266	.end(JSON.stringify({ message: msg, error: error }));
267	},
268	};
269
270	const stacksNodeRpcProxy = createProxyMiddleware(proxyOptions);	207✔
271
272	router.use(stacksNodeRpcProxy);	207✔
273
274	return router;	207✔
275	}

hirosystems / stacks-blockchain-api / 4701730719

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