How do I rate limit and explicitly perform a JSON-RPC batch request?

Question

I noticed in the library README for the experimental web3.js SDK that two of the motivations for the API design are:

• Solana developer ‘epicfaace’ wanted first-class support for automatic time-windowed batching in the RPC transport. Support solana-web3.js batching solana#23628.
• Multiple folks have expressed the need for custom retry logic for failed requests or transactions. fix: double the wait time between each 429 induced retry solana#11811 and Allow disabling retry on HTTP 429 with connection config #1041 attempting to modify retry logic to suit their individual use cases.

However, it is somewhat rather unclear on how i.e. batching or retrying can be implemented with the current experimental SDK.

1. How do I have the HTTP transport automatically merge together specific multiple JSON-RPC payloads into a batch request?
2. How do I synchronously (explicitly) invoke a batch JSON-RPC request with multiple JSON-RPC payloads crafted via. the SDK?
3. How do I pass individual timeouts/retry options for a single/batch JSON-RPC request via. the SDK?

I currently manually perform fetch calls myself to perform explicit batch JSON-RPC requests. This, however, is error-prone as it would require types and validation to be handled by myself (i.e. how do I handle the case that a response field overflows Number.MAX_SAFE_INTEGER?).

export async function getTokenBalances(ownerPublicKey: string) {
  const sol = {
    jsonrpc: "2.0",
    id: 0,
    method: "getAccountInfo",
    params: [
      ownerPublicKey,
      { commitment: "confirmed", encoding: "jsonParsed" },
    ],
  };

  const usdc = {
    jsonrpc: "2.0",
    id: 1,
    method: "getTokenAccountsByOwner",
    params: [
      ownerPublicKey,
      {
        mint: USDC_TOKEN_PUBLIC_KEY,
      },
      {
        commitment: "confirmed",
        encoding: "jsonParsed",
      },
    ],
  };

  const like = {
    jsonrpc: "2.0",
    id: 2,
    method: "getTokenAccountsByOwner",
    params: [
      ownerPublicKey,
      {
        mint: LIKE_TOKEN_PUBLIC_KEY,
      },
      {
        commitment: "confirmed",
        encoding: "jsonParsed",
      },
    ],
  };

  const request = [sol, usdc, like];

  const response = await fetch(SOLANA_SERVER_RPC_URL, {
    method: "POST",
    headers: {
      "Content-Type": "application/json",
    },
    body: JSON.stringify(request),
  });
  if (!response.ok) {
    throw new Error(`RPC request failed: ${response.statusText}`);
  }

  const json = await response.json();
  if (!checkSchema(getTokenBalancesResponse, json)) {
    throw new Error(`Malformed RPC response: ${JSON.stringify(json)}`);
  }

  const balances = {
    sol: BigInt(json[0].result.value?.lamports ?? 0),
    usdc: json[1].result.value.reduce(
      (
        acc,
        {
          account: {
            data: {
              parsed: { info },
            },
          },
        },
      ) => {
        if (info.state !== "initialized") {
          return acc;
        }
        return acc + BigInt(info.tokenAmount.amount);
      },
      BigInt(0),
    ),
    like: json[2].result.value.reduce(
      (
        acc,
        {
          account: {
            data: {
              parsed: { info },
            },
          },
        },
      ) => {
        if (info.state !== "initialized") {
          return acc;
        }
        return acc + BigInt(info.tokenAmount.amount);
      },
      BigInt(0),
    ),
  };

  return balances;
}

Answer 1

With all of these answers, you have to answer the question ‘what requests qualify for inclusion into a batch.’

The obvious criteria is the method name and arguments (eg. only getAccountInfo calls with the same commitment/filters can be batched together). A more controversial criterion is the time delay; is the cutoff for inclusion in a batch measured in ms? Is it based on a single JS runloop? Is it based on the number of requests received? A mix?

For the purpose of this answer, the criterion for inclusion in a batch will be ‘same method, same options, requested in the same runloop.’

Consider the following transport:

import type { RpcTransport } from '@solana/rpc-spec';

type BatchedRequest<T> = {
    readonly params: unknown;
    readonly reject: (reason?: unknown) => void;
    readonly resolve: (value: T) => void;
};

type GetBatchMethodAndOptionsHashFn = (
    payload: unknown,
) => Readonly<{ method: string; optionsHash: string }> | undefined;
type ExecuteBatchFn<T> = (method: string, requests: BatchedRequest<T>[]) => void;

export function getRpcTransportWithBatching<TTransport extends RpcTransport>(
    transport: TTransport,
    getBatchMethodAndOptionsHash: GetBatchMethodAndOptionsHashFn,
    executeBatch: ExecuteBatchFn<unknown>,
): TTransport {
    let batchedRequestsByMethodAndOptionsHash: Record<string, Record<string, BatchedRequest<unknown>[]>> | undefined;
    return async function makeCoalescedHttpRequest<TResponse>(config: Parameters<RpcTransport>[0]): Promise<TResponse> {
        const { payload, signal } = config;
        const methodAndOptionsHash = getBatchMethodAndOptionsHash(payload);
        if (methodAndOptionsHash === undefined) {
            // We have no batch strategy for this request.
            // Resolve it with the underlying transport.
            return await transport(config);
        }
        if (!batchedRequestsByMethodAndOptionsHash) {
            // This is the first time a batch is being created.
            // Schedule the batch for processing at the end of the runloop.
            Promise.resolve().then(() => {
                const currentBatchedRequestsByMethodAndOptionsHash = batchedRequestsByMethodAndOptionsHash;
                batchedRequestsByMethodAndOptionsHash = undefined;
                if (currentBatchedRequestsByMethodAndOptionsHash) {
                    Object.entries(currentBatchedRequestsByMethodAndOptionsHash).forEach(
                        ([method, batchesByOptionsHash]) => {
                            Object.values(batchesByOptionsHash).forEach(batchedRequests => {
                                executeBatch(method, batchedRequests);
                            });
                        },
                    );
                }
            });
            batchedRequestsByMethodAndOptionsHash = {};
        }
        const { method, optionsHash } = methodAndOptionsHash;
        const batchedRequestsByOptionsHash = (batchedRequestsByMethodAndOptionsHash[method] ||= {});
        const batchedRequests = (batchedRequestsByOptionsHash[optionsHash] ||= []);
        const { promise: responsePromise, resolve, reject } = Promise.withResolvers<TResponse>();
        batchedRequests.push({ params: payload.params, reject, resolve });
        if (signal) {
            signal.addEventListener('abort', reject);
        }
        return await (responsePromise as Promise<TResponse>);
    } as TTransport;
}

You could create such a transport with the batching strategy of your choosing:

const batchingTransport = getRpcTransportWithBatching(
    transport,
    function getBatchMethodAndOptionsHash(payload) {
        const [method, options] = payload.params;
        if (method === 'getAccountInfo') {
            return {
                method: 'getMultipleAccounts',
                // Two requests that would result in a fetch with the same commitment/filters
                // should produce the same hash string. Exercise left up to the reader.
                optionsHash: getOptionsHash(method, options),
            };
        }
        // We have no batching strategy for this type of request.
        return undefined;
    },
    function executeBatch(method, requests, transport) {
        if (method === 'getMultipleAccounts') {
            // Collect all of the addresses in the batch.
            const addresses = requests.map(request => request.params[0]);
            // Make a single request for the batch
            transport({ payload: { method, params: [addresses, request[0].params] } })
                .then(({ context, value }) => {
                    value.forEach((accountInfo, ii) => {
                        // Fan the responses out to the individual requests.
                        requests[ii].resolve({context, value: accountInfo});
                    });
                })
                .catch(e => {
                    // If the batch fails, all of the requests fail.
                    requests.forEach(request => request.reject(e));
                });
        }
        throw new Error(`Don't know how to prepare batches for ${method}`);
    },
);

This implementation is missing many necessary features:

1. If an individual request aborts before the batch is prepared, remove it from the list.
2. If all requests abort before the batch is made, delete the pending batch.
3. If all requests abort after the batch request is made but before it resolves, abort the batch request.
4. The option hash function must be aware that two requests with params in different orders, or one request without an option and one with an option that happens to be the default (eg. omitting the commitment vs. setting it to 'confirmed') might need to produce the same hash.
5. In the example given, two requests for the same address with the same options hash should not produce two addresses in the batch; it should produce one that gets fanned out to two callers.
6. In the example given, you need protection against the batches getting too large. When the number of addresses exceeds the maximum allowable by getMultipleAccounts then executeBatch should make multiple requests.

Answer 2

I've just created a library that helps automate the process of collecting requests based on a time window and batch size.

The usage is very simple. Instead of:

    connection.getParsedAccountInfo(account)

you just need to change it to:

    getParsedAccountInBatch(connection, account)

https://github.com/huybuidac/solana-batch-requests/blob/main/packages/solana-batch-requests/src/index.ts#L17