fix(graphql): add abort controller support on interactions endpoint

We use a synchrnoized abort signal for both interaction and state evaluation to ensure we do not continue doing work when we have thrown a timeout error.

src/api/graphql.ts

@@ -27,12 +27,14 @@ import {
 } from 'warp-contracts';
 import { ReadThroughPromiseCache } from '@ardrive/ardrive-promise-cache';
 import winston from 'winston';
+import { EvaluationTimeoutError } from '../errors';
 
 export const MAX_REQUEST_SIZE = 100;
 
 export async function getDeployedContractsByWallet(
  arweave: Arweave,
  params: { address: string },
+ signal?: AbortSignal,
 ): Promise<{ ids: string[] }> {
  const { address } = params;
  let hasNextPage = false;
@@ -71,9 +73,16 @@ export async function getDeployedContractsByWallet(
  }`,
  };
 
+ if (signal?.aborted) {
+ throw new EvaluationTimeoutError();
+ }
+
  const { status, ...response } = await arweave.api.post(
  '/graphql',
  queryObject,
+ {
+ signal,
+ },
  );
  if (status !== 200) {
  throw Error(
@@ -115,6 +124,7 @@ class ContractInteractionsCacheKey {
  public readonly blockHeight: number | undefined,
  public readonly address?: string | undefined,
  public readonly logger?: winston.Logger | undefined,
+ public readonly signal?: AbortSignal | undefined,
  ) {}
 
  toString(): string {
@@ -147,11 +157,13 @@ export async function getWalletInteractionsForContract({
  address,
  blockHeight,
  logger,
+ signal,
 }: {
  arweave: Arweave;
  contractTxId: string;
  address?: string | undefined;
  blockHeight?: number | undefined;
+ signal?: AbortSignal;
  logger?: winston.Logger | undefined;
 }) {
  return contractInteractionsCache.get(
@@ -161,6 +173,7 @@ export async function getWalletInteractionsForContract({
  blockHeight,
  address,
  logger,
+ signal,
  ),
  );
 }
@@ -176,6 +189,7 @@ export async function readThroughToWalletInteractionsForContract(
  address,
  blockHeight: blockHeightFilter,
  logger,
+ signal,
  } = cacheKey;
  logger?.debug('Reading through to wallet interactions for contract...', {
  contractTxId,
@@ -242,9 +256,16 @@ export async function readThroughToWalletInteractionsForContract(
  }`,
  };
 
+ if (signal?.aborted) {
+ throw new EvaluationTimeoutError();
+ }
+
  const { status, data } = await arweave.api.post<GQLResultInterface>(
  '/graphql',
  queryObject,
+ {
+ signal,
+ },
  );
 
  if (status !== 200) {
@@ -309,6 +330,7 @@ export async function readThroughToWalletInteractionsForContract(
 export async function getContractsTransferredToOrControlledByWallet(
  arweave: Arweave,
  params: { address: string },
+ signal?: AbortSignal,
 ): Promise<{ ids: string[] }> {
  const { address } = params;
  let hasNextPage = false;
@@ -406,9 +428,16 @@ export async function getContractsTransferredToOrControlledByWallet(
  }`,
  };
 
+ if (signal?.aborted) {
+ throw new EvaluationTimeoutError();
+ }
+
  const { status, ...response } = await arweave.api.post(
  '/graphql',
  queryObject,
+ {
+ signal,
+ },
  );
  if (status !== 200) {
  throw Error(

src/api/warp.ts

@@ -393,6 +393,7 @@ export async function readThroughToContractReadInteraction(
  .setEvaluationOptions(evaluationOptions);
 
  // set cached value for multiple requests during initial promise
+ // TODO: add abort signal when view state supports it
  const readInteractionPromise = contract.viewState({
  function: functionName,
  ...input,

src/errors.ts

@@ -15,7 +15,6 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
-// TODO: we could put a prometheus metric here to help fine tune what our evaluation limit should be
 export class BaseError extends Error {
  super(message: string) {
  this.message = message;
@@ -24,7 +23,8 @@ export class BaseError extends Error {
 }
 export class EvaluationTimeoutError extends BaseError {
  constructor() {
- super(`Evaluation timed out for contract.`);
+ super(`Evaluation timed out.`);
+ // TODO: we could put a prometheus metric here to help fine tune what our evaluation limit should be
  }
 }
 export class EvaluationError extends BaseError {}

src/routes/contract.ts

@@ -25,6 +25,7 @@ import { getContractReadInteraction, getContractState } from '../api/warp';
 import { getWalletInteractionsForContract } from '../api/graphql';
 import { NotFoundError } from '../errors';
 import { mismatchedInteractionCount } from '../metrics';
+import { DEFAULT_STATE_EVALUATION_TIMEOUT_MS } from '../constants';
 
 export async function contractHandler(ctx: KoaContext) {
  const {
@@ -81,6 +82,8 @@ export async function contractInteractionsHandler(ctx: KoaContext) {
  * TODO: add a read through promise cache here that uses the following logic as the resulting promise. The cache key should contain the contractTxId, sortKey, blockHeight, address, page, and pageSize.
  */
 
+ // use a synchronized abort signal to stop execution of both when either times out
+ const abortSignal = AbortSignal.timeout(DEFAULT_STATE_EVALUATION_TIMEOUT_MS);
  const [
  { validity, errorMessages, evaluationOptions, sortKey: evaluatedSortKey },
  { interactions },
@@ -91,13 +94,15 @@ export async function contractInteractionsHandler(ctx: KoaContext) {
  logger,
  sortKey: requestedSortKey,
  blockHeight: requestedBlockHeight,
+ signal: abortSignal,
  }),
  getWalletInteractionsForContract({
  arweave,
  address,
  contractTxId,
  blockHeight: requestedBlockHeight,
  logger,
+ signal: abortSignal,
  }),
  ]);