better tokenpool downloading

README.md

@@ -34,6 +34,8 @@ ALLOWLIST_DB_PASSWORD=allowlist
 ALLOWLIST_ETHERSCAN_API_KEY=<your-etherscan-api-key>
 ALLOWLIST_SEIZE_API_PATH=<seize-api-endpoint> (don't put a slash in the end. https://api.6529.io/api for example)
 ALLOWLIST_SEIZE_API_KEY=<seize-api-key> (can omit if using only public endpoints)
+ALLOWLIST_SEIZE_METADATA_TIMEOUT_MS=10000 (optional)
+ALLOWLIST_ARWEAVE_DOWNLOAD_TIMEOUT_MS=30000 (optional)
 ```
 
 To install app dependencies run `yarn`
@@ -65,4 +67,4 @@ In production the app is ran in 3 lambas:
 
 1. API lambda - Serves all API requests (entrypoint: `src/api-lambda.ts/handle`)
 2. Worker lambda - Does the actual final allowlist creation (entrypoint: `src/worker-lambda.ts/handle`)
-3. Tokenpool downloader lambda - Helps to get aggregated tokenpool data needed for worker lambda (entrypoint: `src/token-pool-downloader-lambda.ts/handle`)
+3. Tokenpool downloader lambda - Helps to get aggregated tokenpool data needed for worker lambda (entrypoint: `src/token-pool-downloader-lambda.ts/handle`)

src/allowlist-lib/allowlist-lib-execution-context.service.ts

@@ -0,0 +1,35 @@
+import { Injectable } from '@nestjs/common';
+import { AsyncLocalStorage } from 'async_hooks';
+
+export interface AllowlistLibExecutionContext {
+  readonly tokenPoolId: string;
+  readonly contract: string;
+  readonly blockNo: number;
+  readonly consolidateBlockNo: number | null;
+}
+
+@Injectable()
+export class AllowlistLibExecutionContextService {
+  private readonly storage =
+    new AsyncLocalStorage<AllowlistLibExecutionContext>();
+
+  run<T>(
+    context: AllowlistLibExecutionContext,
+    callback: () => Promise<T> | T,
+  ): Promise<T> | T {
+    return this.storage.run(context, callback);
+  }
+
+  getLogPrefix(): string {
+    const context = this.storage.getStore();
+    if (!context) {
+      return '';
+    }
+    return [
+      `tokenPoolId=${context.tokenPoolId}`,
+      `contract=${context.contract}`,
+      `blockNo=${context.blockNo}`,
+      `consolidateBlockNo=${context.consolidateBlockNo ?? 'null'}`,
+    ].join(' ');
+  }
+}

src/allowlist-lib/allowlist-lib-log-listener.service.spec.ts

@@ -0,0 +1,27 @@
+import { Logger } from '@nestjs/common';
+import { AllowlistLibExecutionContextService } from './allowlist-lib-execution-context.service';
+import { AllowlistLibLogListener } from './allowlist-lib-log-listener.service';
+
+describe('AllowlistLibLogListener', () => {
+  it('prefixes messages with token pool context when present', () => {
+    const executionContext = new AllowlistLibExecutionContextService();
+    const listener = new AllowlistLibLogListener(executionContext);
+    const logSpy = jest.spyOn(Logger.prototype, 'log').mockImplementation();
+
+    executionContext.run(
+      {
+        tokenPoolId: 'pool-1',
+        contract: '0xabc',
+        blockNo: 42,
+        consolidateBlockNo: 99,
+      },
+      () => listener.info('Downloading from URL: https://arweave.net/example'),
+    );
+
+    expect(logSpy).toHaveBeenCalledWith(
+      '[tokenPoolId=pool-1 contract=0xabc blockNo=42 consolidateBlockNo=99] Downloading from URL: https://arweave.net/example',
+    );
+
+    logSpy.mockRestore();
+  });
+});

src/allowlist-lib/allowlist-lib-log-listener.service.ts

@@ -1,23 +1,36 @@
 import { Injectable, Logger } from '@nestjs/common';
 import { LogListener } from '@6529-collections/allowlist-lib/logging/logging-emitter';
+import { AllowlistLibExecutionContextService } from './allowlist-lib-execution-context.service';
 
 const nestJsLogger = new Logger('allowlist-lib');
 
 @Injectable()
 export class AllowlistLibLogListener implements LogListener {
+  constructor(
+    private readonly executionContext: AllowlistLibExecutionContextService,
+  ) {}
+
   debug(message: string): void {
-    nestJsLogger.debug(message);
+    nestJsLogger.debug(this.withContext(message));
   }
 
   error(message: string): void {
-    nestJsLogger.error(message);
+    nestJsLogger.error(this.withContext(message));
   }
 
   info(message: string): void {
-    nestJsLogger.log(message);
+    nestJsLogger.log(this.withContext(message));
   }
 
   warn(message: string): void {
-    nestJsLogger.warn(message);
+    nestJsLogger.warn(this.withContext(message));
+  }
+
+  private withContext(message: string) {
+    const prefix = this.executionContext.getLogPrefix();
+    if (!prefix) {
+      return message;
+    }
+    return `[${prefix}] ${message}`;
   }
 }

src/allowlist-lib/allowlist-lib-seize-timeout-patch.spec.ts

@@ -0,0 +1,141 @@
+import { AllowlistCreator } from '@6529-collections/allowlist-lib/allowlist/allowlist-creator';
+import {
+  LoggerFactory,
+  LogListener,
+} from '@6529-collections/allowlist-lib/logging/logging-emitter';
+import { SeizeApi } from '@6529-collections/allowlist-lib/services/seize/seize.api';
+import axios from 'axios';
+import {
+  parseTimeoutMs,
+  patchAllowlistCreatorSeizeApi,
+} from './allowlist-lib-seize-timeout-patch';
+
+const minimalTdhUploadContents =
+  'wallet,ens,consolidation_key,consolidation_display,block,date,total_balance,boosted_tdh,tdh_rank,tdh,tdh__raw,boost,memes_balance,unique_memes,memes_cards_sets,memes_cards_sets_minus1,memes_cards_sets_minus2,genesis,nakamoto,boosted_memes_tdh,memes_tdh,memes_tdh__raw,tdh_rank_memes,memes,gradients_balance,boosted_gradients_tdh,gradients_tdh,gradients_tdh__raw,tdh_rank_gradients,gradients,nextgen_balance,boosted_nextgen_tdh,nextgen_tdh,nextgen_tdh__raw,nextgen\n' +
+  '0xAbC,test.eth,key,display,1,20240101,0,0,1,0,0,1,0,0,0,0,0,0,0,0,0,0,1,[],0,0,0,0,1,[],0,0,0,0,[]\n';
+
+class TestLogListener implements LogListener {
+  readonly infoMessages: string[] = [];
+  readonly errorMessages: string[] = [];
+
+  debug = (): void => {};
+
+  error = (message: string): void => {
+    this.errorMessages.push(message);
+  };
+
+  info = (message: string): void => {
+    this.infoMessages.push(message);
+  };
+
+  warn = (): void => {};
+}
+
+describe('patchAllowlistCreatorSeizeApi', () => {
+  const axiosGetSpy = jest.spyOn(axios, 'get');
+  const axiosIsAxiosErrorSpy = jest.spyOn(axios, 'isAxiosError');
+
+  beforeEach(() => {
+    jest.resetAllMocks();
+    axiosIsAxiosErrorSpy.mockImplementation(
+      (value: unknown): value is Error =>
+        !!value && typeof value === 'object' && 'isAxiosError' in value,
+    );
+  });
+
+  afterAll(() => {
+    axiosGetSpy.mockRestore();
+    axiosIsAxiosErrorSpy.mockRestore();
+  });
+
+  it('falls back to the configured default timeout when env is invalid', () => {
+    expect(parseTimeoutMs(undefined, 10000)).toBe(10000);
+    expect(parseTimeoutMs('not-a-number', 10000)).toBe(10000);
+    expect(parseTimeoutMs('5000', 10000)).toBe(5000);
+  });
+
+  it('surfaces Seize metadata timeouts with the configured timeout value', async () => {
+    const listener = new TestLogListener();
+    const seizeApi = new SeizeApi({} as any, 'https://www.example.com/api');
+    const allowlistCreator = { seizeApi } as unknown as AllowlistCreator;
+
+    patchAllowlistCreatorSeizeApi({
+      allowlistCreator,
+      loggerFactory: new LoggerFactory(listener),
+      seizeMetadataTimeoutMs: 10,
+      arweaveDownloadTimeoutMs: 25,
+    });
+
+    axiosGetSpy.mockRejectedValueOnce({
+      isAxiosError: true,
+      code: 'ECONNABORTED',
+      message: 'timeout of 10ms exceeded',
+    });
+
+    await expect(
+      (seizeApi as any).getDataForBlock({ path: '/uploads', blockId: 123 }),
+    ).rejects.toThrow(
+      'Seize metadata fetch timed out after 10ms for https://www.example.com/api/uploads?block=123&page_size=1',
+    );
+    expect(listener.infoMessages).toEqual(
+      expect.arrayContaining([
+        expect.stringContaining(
+          'Fetching Seize metadata from https://www.example.com/api/uploads?block=123&page_size=1',
+        ),
+      ]),
+    );
+  });
+
+  it('logs and falls back to the next Arweave gateway after a timeout', async () => {
+    const listener = new TestLogListener();
+    const seizeApi = new SeizeApi({} as any, 'https://www.example.com/api');
+    const allowlistCreator = { seizeApi } as unknown as AllowlistCreator;
+
+    patchAllowlistCreatorSeizeApi({
+      allowlistCreator,
+      loggerFactory: new LoggerFactory(listener),
+      seizeMetadataTimeoutMs: 10,
+      arweaveDownloadTimeoutMs: 25,
+    });
+
+    axiosGetSpy
+      .mockResolvedValueOnce({
+        data: {
+          data: [
+            {
+              block: 17531454,
+              url: 'https://arweave.net/abc123',
+            },
+          ],
+        },
+      })
+      .mockRejectedValueOnce({
+        isAxiosError: true,
+        code: 'ECONNABORTED',
+        message: 'timeout of 25ms exceeded',
+      })
+      .mockResolvedValueOnce({
+        data: minimalTdhUploadContents,
+        headers: { 'content-type': 'text/csv; charset=utf-8' },
+      });
+
+    const uploads = await seizeApi.getUploadsForBlock(17531454);
+
+    expect(uploads).toHaveLength(1);
+    expect(uploads[0].wallet).toBe('0xabc');
+    expect(listener.infoMessages).toEqual(
+      expect.arrayContaining([
+        expect.stringContaining(
+          'Downloading from URL: https://gateway.arweave.net/abc123',
+        ),
+      ]),
+    );
+    expect(listener.errorMessages).toEqual(
+      expect.arrayContaining([
+        expect.stringContaining(
+          'Failed to download from URL: https://arweave.net/abc123 because of error: Arweave CSV download timed out after 25ms for https://arweave.net/abc123',
+        ),
+      ]),
+    );
+  });
+});