diff --git a/src/application/index.ts b/src/application/index.ts index 3061ddb..0506c2f 100644 --- a/src/application/index.ts +++ b/src/application/index.ts @@ -20,13 +20,14 @@ import { SimulationScheduler } from '../simulation/scheduler.js'; import { SignalNormalizer } from '../signals/index.js'; import { IWalletWatcher } from '../domain/interfaces.js'; import { RawWalletTransaction } from '../domain/models.js'; -import { fetchTransactionWithRetry } from '../solana/fetcher.js'; +import { BoundedTransactionQueue, BackpressureDropError } from '../solana/fetcher.js'; export class Application { public isShuttingDown = false; public simulationIntervalId: NodeJS.Timeout | null = null; public watcher: IWalletWatcher | null = null; public metricsProvider: InfluxMetricsProvider | MockMetricsProvider | null = null; + public transactionQueue: BoundedTransactionQueue | null = null; async bootstrap(): Promise { const config = loadConfig(); @@ -131,6 +132,14 @@ export class Application { config.SOLANA_RPC_ENDPOINT, config.SOLANA_WSS_ENDPOINT, ); + this.transactionQueue = new BoundedTransactionQueue( + (this.watcher as any).connection || new (await import('@solana/web3.js')).Connection(config.SOLANA_RPC_ENDPOINT), + config.SOLANA_QUEUE_MAX_CONCURRENCY, + config.SOLANA_QUEUE_MIN_DELAY_MS, + config.SOLANA_QUEUE_MAX_SIZE, + this.metricsProvider || undefined, + ); + this.transactionQueue.start(); } else { this.watcher = new MockWalletWatcher(watchedAddresses); } @@ -141,6 +150,14 @@ export class Application { const detectedAt = observation.timestamp || new Date(); if (config.WALLET_WATCHER_MODE === 'real') { + // Pre-filtering failed transactions: + // WebSocket subscriptions (like logs) may provide an "err" field. If err is present, filter it! + const wsPayload = (observation as any).rawPayload || {}; + if (wsPayload.err) { + logger.info(`[RealMode] Pre-filtered failed transaction signature ${observation.signature} (logs.err is not null).`, { correlationId }); + return; + } + logger.info( `[RealMode] Activity detected for wallet ${observation.walletAddress}. Signature: ${observation.signature}, Slot: ${observation.slot}`, { correlationId }, @@ -169,72 +186,110 @@ export class Application { observation.walletAddress, ); - // Fetch transaction - const fetchStartTime = Date.now(); - await journalService.log(correlationId, 'raw_transaction_fetch_started', { - signature: observation.signature, - }); - - let txPayload: any; - try { - const connection = (this.watcher as any).connection; - if (!connection) { - throw new Error('Connection not initialized on RealWalletWatcher'); - } - txPayload = await fetchTransactionWithRetry(connection, observation.signature); - } catch (error) { - logger.error(`[RealMode] Failed to fetch transaction ${observation.signature}`, error); - await journalService.log(correlationId, 'raw_transaction_fetch_failed', { - signature: observation.signature, - error: error instanceof Error ? error.message : String(error), - }); + // Push item onto the bounded queue for non-blocking execution + if (!this.transactionQueue) { + logger.error('[RealMode] Transaction queue not initialized!'); return; } - const fetchEndTime = Date.now(); - const fetchDuration = fetchEndTime - fetchStartTime; - this.metricsProvider?.recordQuoteLatency(fetchDuration, 'rpc_fetch'); - - await journalService.log(correlationId, 'raw_transaction_fetched', { + const queued = this.transactionQueue.push({ signature: observation.signature, - slot: txPayload.slot, - blockTime: txPayload.blockTime, + walletAddress: observation.walletAddress, + slot: observation.slot, + detectedAt, + correlationId, + onSuccess: async (txPayload) => { + await journalService.log(correlationId, 'raw_transaction_fetched', { + signature: observation.signature, + slot: txPayload.slot, + blockTime: txPayload.blockTime, + }); + + // Calculate balance changes to assess transaction relevance + const { isTransactionRelevant } = await import('../solana/parser.js'); + const relevant = isTransactionRelevant(txPayload, observation.walletAddress); + + this.metricsProvider?.recordTransactionRelevance(relevant); + + if (!relevant) { + logger.info(`[RealMode] Classification: IRRELEVANT transaction ${observation.signature} for wallet ${observation.walletAddress}. Excluding from persistence.`, { correlationId }); + await journalService.log(correlationId, 'transaction_classified_irrelevant', { + signature: observation.signature, + walletAddress: observation.walletAddress, + }); + + // If configured to skip storing irrelevant transactions, stop here! + if (!config.STORE_IRRELEVANT_RAW_TRANSACTIONS) { + return; + } + } else { + logger.info(`[RealMode] Classification: RELEVANT transaction ${observation.signature} detected.`, { correlationId }); + await journalService.log(correlationId, 'transaction_classified_relevant', { + signature: observation.signature, + walletAddress: observation.walletAddress, + }); + } + + // Persist raw transaction + try { + const blockTime = txPayload.blockTime ? new Date(txPayload.blockTime * 1000) : null; + + if (blockTime) { + this.metricsProvider?.recordWalletDetectionLatency( + Date.now() - blockTime.getTime(), + observation.walletAddress, + ); + } + + const rawTx: RawWalletTransaction = { + transactionSignature: observation.signature, + sourceWallet: observation.walletAddress, + slot: observation.slot, + blockTime, + detectedAt, + rawPayload: txPayload, + parserVersion: 'ingestion_v1', + }; + + await rawTxRepo.saveTransaction(rawTx); + + await journalService.log(correlationId, 'raw_transaction_persisted', { + signature: observation.signature, + slot: observation.slot, + }); + + logger.info(`[RealMode] Raw transaction ${observation.signature} successfully persisted and pipeline STOPPED.`, { correlationId }); + } catch (error) { + logger.error(`[RealMode] Failed to persist raw transaction ${observation.signature}`, error); + } + }, + onFailure: async (error) => { + const isBackpressure = error instanceof BackpressureDropError || error.name === 'BackpressureDropError'; + if (isBackpressure) { + const bpError = error as BackpressureDropError; + logger.warn(`[RealMode] Transaction dropped due to backpressure: signature ${bpError.signature} for wallet ${bpError.walletAddress}. Depth: ${bpError.queueDepth}, Age: ${bpError.ageMs}ms`); + await journalService.log(correlationId, 'transaction_dropped_backpressure', { + signature: bpError.signature, + walletAddress: bpError.walletAddress, + queueDepth: bpError.queueDepth, + ageMs: bpError.ageMs, + }); + } else { + logger.error(`[RealMode] Failed to fetch transaction ${observation.signature}`, error); + await journalService.log(correlationId, 'raw_transaction_fetch_failed', { + signature: observation.signature, + error: error.message, + }); + } + }, }); - // Persist raw transaction - try { - const blockTime = txPayload.blockTime ? new Date(txPayload.blockTime * 1000) : null; - - if (blockTime) { - this.metricsProvider?.recordWalletDetectionLatency( - Date.now() - blockTime.getTime(), - observation.walletAddress, - ); - } - - const rawTx: RawWalletTransaction = { - transactionSignature: observation.signature, - sourceWallet: observation.walletAddress, - slot: observation.slot, - blockTime, - detectedAt, - rawPayload: txPayload, - parserVersion: 'ingestion_v1', - }; - - await rawTxRepo.saveTransaction(rawTx); - - const totalDuration = Date.now() - detectedAt.getTime(); - this.metricsProvider?.recordSignalProcessingLatency(totalDuration); - - await journalService.log(correlationId, 'raw_transaction_persisted', { + if (!queued) { + logger.warn(`[RealMode] Signature ${observation.signature} was dropped due to queue backpressure.`, { correlationId }); + await journalService.log(correlationId, 'raw_transaction_fetch_failed', { signature: observation.signature, - slot: observation.slot, + error: 'Queue backpressure: item dropped', }); - - logger.info(`[RealMode] Raw transaction ${observation.signature} successfully persisted and pipeline STOPPED.`, { correlationId }); - } catch (error) { - logger.error(`[RealMode] Failed to persist raw transaction ${observation.signature}`, error); } // STOP (No strategy evaluation, scheduler or copy signal creation in real mode!) @@ -310,14 +365,16 @@ export class Application { // 7. Start on-chain watcher await this.watcher.startWatching(); - // 8. Run simulator poller non-blockingly every 500ms - this.simulationIntervalId = setInterval(async () => { - try { - await simulationScheduler.processPending(new Date()); - } catch (err) { - logger.error('Error in simulation engine processing loop', err); - } - }, 500); + // 8. Run simulator poller non-blockingly every 500ms (only if NOT in real mode) + if (config.WALLET_WATCHER_MODE !== 'real') { + this.simulationIntervalId = setInterval(async () => { + try { + await simulationScheduler.processPending(new Date()); + } catch (err) { + logger.error('Error in simulation engine processing loop', err); + } + }, 500); + } logger.info('Bot system components started and fully operational.'); } @@ -333,6 +390,11 @@ export class Application { this.simulationIntervalId = null; } + if (this.transactionQueue) { + this.transactionQueue.stop(); + this.transactionQueue = null; + } + if (this.watcher) { await this.watcher.stopWatching(); } diff --git a/src/config/index.ts b/src/config/index.ts index ef586e2..fd8ece5 100644 --- a/src/config/index.ts +++ b/src/config/index.ts @@ -40,6 +40,22 @@ const configSchema = z.object({ .transform((val) => (typeof val === 'number' ? val : parseFloat(val))) .default(0.05), WALLET_WATCHER_MODE: z.enum(['real', 'mock']).default('real'), + SOLANA_QUEUE_MAX_CONCURRENCY: z + .union([z.string(), z.number()]) + .transform((val) => (typeof val === 'number' ? val : parseInt(val, 10))) + .default(2), + SOLANA_QUEUE_MIN_DELAY_MS: z + .union([z.string(), z.number()]) + .transform((val) => (typeof val === 'number' ? val : parseInt(val, 10))) + .default(500), + SOLANA_QUEUE_MAX_SIZE: z + .union([z.string(), z.number()]) + .transform((val) => (typeof val === 'number' ? val : parseInt(val, 10))) + .default(100), + STORE_IRRELEVANT_RAW_TRANSACTIONS: z + .union([z.string(), z.boolean()]) + .transform((val) => (typeof val === 'boolean' ? val : val === 'true')) + .default(false), CONFIGURATION_HASH: z.string().default(''), }); diff --git a/src/domain/interfaces.ts b/src/domain/interfaces.ts index 3c4fe8e..fb9edf3 100644 --- a/src/domain/interfaces.ts +++ b/src/domain/interfaces.ts @@ -130,4 +130,17 @@ export interface IMetricsProvider { recordPortfolioBalance(balanceSol: number, scenario: number): void; recordTradePnl(pnlSol: number, scenario: number, tokenMint: string): void; recordExcursion(mfeSol: number, maeSol: number, scenario: number, tokenMint: string): void; + // Queue & rate limiting metrics + recordQueueDepth(depth: number): void; + recordQueueRetries(retries: number, signature: string): void; + recordQueue429s(signature: string): void; + // Extended high-performance queue & relevance metrics + recordWalletEventReceived(wallet: string): void; + recordWalletQueueDepth(wallet: string, depth: number): void; + recordOldestQueuedAge(wallet: string, ageMs: number): void; + recordDroppedOldestCount(wallet: string): void; + recordTransactionRelevance(relevant: boolean): void; + recordRpcRequestRate(requestsPerSecond: number): void; + recordQueueProcessed(wallet: string): void; + recordRpcFetchError(wallet: string): void; } diff --git a/src/metrics/index.ts b/src/metrics/index.ts index 7121286..9f073dd 100644 --- a/src/metrics/index.ts +++ b/src/metrics/index.ts @@ -91,6 +91,61 @@ export class InfluxMetricsProvider implements IMetricsProvider { this.writePoint(p); } + recordQueueDepth(depth: number): void { + const p = new Point('queue_depth').intField('depth', depth); + this.writePoint(p); + } + + recordQueueRetries(retries: number, signature: string): void { + const p = new Point('queue_retries').tag('signature', signature).intField('retries', retries); + this.writePoint(p); + } + + recordQueue429s(signature: string): void { + const p = new Point('queue_429s').tag('signature', signature).intField('count', 1); + this.writePoint(p); + } + + recordWalletEventReceived(wallet: string): void { + const p = new Point('wallet_event_received').tag('wallet', wallet).intField('count', 1); + this.writePoint(p); + } + + recordWalletQueueDepth(wallet: string, depth: number): void { + const p = new Point('wallet_queue_depth').tag('wallet', wallet).intField('depth', depth); + this.writePoint(p); + } + + recordOldestQueuedAge(wallet: string, ageMs: number): void { + const p = new Point('oldest_queued_age').tag('wallet', wallet).floatField('age_ms', ageMs); + this.writePoint(p); + } + + recordDroppedOldestCount(wallet: string): void { + const p = new Point('dropped_oldest_count').tag('wallet', wallet).intField('count', 1); + this.writePoint(p); + } + + recordTransactionRelevance(relevant: boolean): void { + const p = new Point('transaction_relevance').tag('relevant', String(relevant)).intField('count', 1); + this.writePoint(p); + } + + recordRpcRequestRate(requestsPerSecond: number): void { + const p = new Point('rpc_request_rate').floatField('req_per_sec', requestsPerSecond); + this.writePoint(p); + } + + recordQueueProcessed(wallet: string): void { + const p = new Point('queue_processed').tag('wallet', wallet).intField('count', 1); + this.writePoint(p); + } + + recordRpcFetchError(wallet: string): void { + const p = new Point('rpc_fetch_error').tag('wallet', wallet).intField('count', 1); + this.writePoint(p); + } + async close(): Promise { if (this.writeApi) { logger.info('Closing InfluxDB metrics writer...'); @@ -114,4 +169,15 @@ export class MockMetricsProvider implements IMetricsProvider { recordPortfolioBalance(_balanceSol: number, _scenario: number): void {} recordTradePnl(_pnlSol: number, _scenario: number, _tokenMint: string): void {} recordExcursion(_mfeSol: number, _maeSol: number, _scenario: number, _tokenMint: string): void {} + recordQueueDepth(_depth: number): void {} + recordQueueRetries(_retries: number, _signature: string): void {} + recordQueue429s(_signature: string): void {} + recordWalletEventReceived(_wallet: string): void {} + recordWalletQueueDepth(_wallet: string, _depth: number): void {} + recordOldestQueuedAge(_wallet: string, _ageMs: number): void {} + recordDroppedOldestCount(_wallet: string): void {} + recordTransactionRelevance(_relevant: boolean): void {} + recordRpcRequestRate(_requestsPerSecond: number): void {} + recordQueueProcessed(_wallet: string): void {} + recordRpcFetchError(_wallet: string): void {} } diff --git a/src/solana/fetcher.ts b/src/solana/fetcher.ts index 27b09af..274fde8 100644 --- a/src/solana/fetcher.ts +++ b/src/solana/fetcher.ts @@ -1,12 +1,14 @@ import { Connection, VersionedTransactionResponse } from '@solana/web3.js'; import { logger } from '../logging/index.js'; +import { IMetricsProvider } from '../domain/interfaces.js'; export async function fetchTransactionWithRetry( connection: Connection, signature: string, maxRetries = 5, initialDelayMs = 500, - maxDelayMs = 4000, + maxDelayMs = 8000, + metricsProvider?: IMetricsProvider, ): Promise { let attempt = 0; let delay = initialDelayMs; @@ -24,8 +26,27 @@ export async function fetchTransactionWithRetry( } logger.warn(`[TransactionFetcher] Transaction ${signature} not found on attempt ${attempt + 1}. Retrying...`); - } catch (err) { - logger.error(`[TransactionFetcher] Error fetching transaction ${signature} on attempt ${attempt + 1}`, err); + } catch (err: any) { + const isRateLimit = err && (err.status === 429 || String(err.message).includes('429') || String(err).includes('429')); + if (isRateLimit) { + logger.error(`[TransactionFetcher] HTTP 429 Rate Limit encountered for ${signature} on attempt ${attempt + 1}`); + if (metricsProvider) { + metricsProvider.recordQueue429s(signature); + } + // Use exponential backoff with jitter for HTTP 429 + const jitter = Math.random() * 200; + const rateLimitDelay = Math.min(delay * 3 + jitter, maxDelayMs * 2); + logger.warn(`[TransactionFetcher] Rate limit delay: waiting ${rateLimitDelay.toFixed(0)}ms before retry`); + await new Promise((resolve) => setTimeout(resolve, rateLimitDelay)); + attempt++; + if (metricsProvider) { + metricsProvider.recordQueueRetries(attempt, signature); + } + delay = Math.min(delay * 2, maxDelayMs); + continue; + } else { + logger.error(`[TransactionFetcher] Error fetching transaction ${signature} on attempt ${attempt + 1}`, err); + } } attempt++; @@ -33,6 +54,10 @@ export async function fetchTransactionWithRetry( break; } + if (metricsProvider) { + metricsProvider.recordQueueRetries(attempt, signature); + } + logger.debug(`[TransactionFetcher] Waiting ${delay}ms before next retry for ${signature}...`); await new Promise((resolve) => setTimeout(resolve, delay)); delay = Math.min(delay * 2, maxDelayMs); @@ -40,3 +65,245 @@ export async function fetchTransactionWithRetry( throw new Error(`Transaction ${signature} could not be fetched after ${maxRetries} attempts.`); } + +export interface QueueItem { + signature: string; + walletAddress: string; + slot: number; + detectedAt: Date; + correlationId: string; + onSuccess: (tx: VersionedTransactionResponse) => Promise; + onFailure: (error: Error) => Promise; +} + +export class BackpressureDropError extends Error { + public walletAddress: string; + public signature: string; + public queueDepth: number; + public ageMs: number; + + constructor(message: string, walletAddress: string, signature: string, queueDepth: number, ageMs: number) { + super(message); + this.name = 'BackpressureDropError'; + this.walletAddress = walletAddress; + this.signature = signature; + this.queueDepth = queueDepth; + this.ageMs = ageMs; + this.stack = undefined; + } +} + +export class BoundedTransactionQueue { + private walletQueues = new Map(); + private walletOrder: string[] = []; // for round-robin scheduling + private activeCount = 0; + private running = false; + private lastRequestTime = 0; + private requestCountLastSecond = 0; + private rateSecondTimer: NodeJS.Timeout | null = null; + private rateLimitTimeout: NodeJS.Timeout | null = null; + + constructor( + private connection: Connection, + private maxConcurrency: number, + private minDelayMs: number, + private maxQueueSize: number, // Bounded limit PER wallet queue + private metricsProvider?: IMetricsProvider, + ) {} + + public start(): void { + if (this.running) return; + this.running = true; + logger.info(`[BoundedTransactionQueue] Started. Concurrency: ${this.maxConcurrency}, MinDelay: ${this.minDelayMs}ms, MaxQueueSize (Per-Wallet): ${this.maxQueueSize}`); + + this.rateSecondTimer = setInterval(() => { + if (this.metricsProvider) { + this.metricsProvider.recordRpcRequestRate(this.requestCountLastSecond); + } + this.requestCountLastSecond = 0; + }, 1000); + } + + public stop(): void { + this.running = false; + if (this.rateSecondTimer) { + clearInterval(this.rateSecondTimer); + this.rateSecondTimer = null; + } + if (this.rateLimitTimeout) { + clearTimeout(this.rateLimitTimeout); + this.rateLimitTimeout = null; + } + logger.info('[BoundedTransactionQueue] Stopped accepting new items. Clearing all per-wallet queues.'); + this.walletQueues.clear(); + this.walletOrder = []; + } + + public getDepth(walletAddress?: string): number { + if (walletAddress) { + return this.walletQueues.get(walletAddress)?.length || 0; + } + let total = 0; + for (const q of this.walletQueues.values()) { + total += q.length; + } + return total; + } + + public getActiveCount(): number { + return this.activeCount; + } + + public push(item: QueueItem): boolean { + if (!this.running) { + logger.warn(`[BoundedTransactionQueue] Rejected signature ${item.signature} because queue is not running`); + item.onFailure(new Error('Queue is not running')).catch(() => {}); + return false; + } + + if (this.metricsProvider) { + this.metricsProvider.recordWalletEventReceived(item.walletAddress); + } + + let wQueue = this.walletQueues.get(item.walletAddress); + if (!wQueue) { + wQueue = []; + this.walletQueues.set(item.walletAddress, wQueue); + } + + // Check bounded size backpressure per-wallet queue + if (wQueue.length >= this.maxQueueSize) { + // BACKPRESSURE policy: favor recent transactions! + // Drop the OLDEST queued item instead of blindly dropping the newest + const dropped = wQueue.shift(); // removes oldest + if (dropped) { + const ageMs = Date.now() - dropped.detectedAt.getTime(); + logger.warn(`[BoundedTransactionQueue] BACKPRESSURE (Wallet ${item.walletAddress}): queue full. Dropped oldest signature: ${dropped.signature}. Appending newest: ${item.signature}`); + if (this.metricsProvider) { + this.metricsProvider.recordDroppedOldestCount(item.walletAddress); + } + dropped.onFailure(new BackpressureDropError( + 'Dropped from queue due to backpressure', + dropped.walletAddress, + dropped.signature, + wQueue.length + 1, // depth when drop occurred + ageMs + )).catch(() => {}); + } + } + + wQueue.push(item); + + // Track oldest queued item age in metrics + if (this.metricsProvider && wQueue.length > 0 && wQueue[0]) { + this.metricsProvider.recordOldestQueuedAge(item.walletAddress, Date.now() - wQueue[0].detectedAt.getTime()); + this.metricsProvider.recordWalletQueueDepth(item.walletAddress, wQueue.length); + this.metricsProvider.recordQueueDepth(this.getDepth()); + } + + // Ensure wallet is in round-robin scheduler order + if (!this.walletOrder.includes(item.walletAddress)) { + this.walletOrder.push(item.walletAddress); + } + + logger.debug(`[BoundedTransactionQueue] Queued signature ${item.signature} for wallet ${item.walletAddress}. Depth: ${wQueue.length}`); + + // Trigger process next non-blockingly + this.processNext(); + return true; + } + + private async processNext(): Promise { + if (!this.running) return; + if (this.activeCount >= this.maxConcurrency) return; + + // Filter out empty queues from round robin orders + this.walletOrder = this.walletOrder.filter(addr => { + const q = this.walletQueues.get(addr); + return q && q.length > 0; + }); + + if (this.walletOrder.length === 0) return; + + // Enforcement of rate limiting delay + const now = Date.now(); + const timeSinceLast = now - this.lastRequestTime; + if (timeSinceLast < this.minDelayMs) { + const waitTime = this.minDelayMs - timeSinceLast; + if (!this.rateLimitTimeout) { + logger.debug(`[BoundedTransactionQueue] Global rate limit: waiting ${waitTime}ms`); + this.rateLimitTimeout = setTimeout(() => { + this.rateLimitTimeout = null; + this.processNext(); + }, waitTime); + } + return; + } + + // Fair round-robin: pull the first wallet address from scheduling list + const walletAddr = this.walletOrder.shift(); + if (!walletAddr) return; + + const wQueue = this.walletQueues.get(walletAddr); + if (!wQueue || wQueue.length === 0) return; + + const item = wQueue.shift(); + if (!item) return; + + // Re-queue the wallet address at the end of the order if it still has pending items + if (wQueue.length > 0) { + this.walletOrder.push(walletAddr); + } + + if (this.metricsProvider) { + this.metricsProvider.recordWalletQueueDepth(walletAddr, wQueue.length); + this.metricsProvider.recordQueueDepth(this.getDepth()); + } + + this.activeCount++; + this.lastRequestTime = Date.now(); + this.requestCountLastSecond++; + + logger.debug(`[BoundedTransactionQueue] Executing item ${item.signature} from fair wallet ${walletAddr}. Active: ${this.activeCount}/${this.maxConcurrency}`); + + // Asynchronously trigger fetch + this.executeFetch(item); + + // Concurrently try triggering others + this.processNext(); + } + + private async executeFetch(item: QueueItem): Promise { + const fetchStartTime = Date.now(); + try { + const tx = await fetchTransactionWithRetry( + this.connection, + item.signature, + 5, + 500, + 4000, + this.metricsProvider, + ); + + const fetchEndTime = Date.now(); + const fetchDuration = fetchEndTime - fetchStartTime; + + if (this.metricsProvider) { + this.metricsProvider.recordQuoteLatency(fetchDuration, 'rpc_fetch'); + this.metricsProvider.recordQueueProcessed(item.walletAddress); + } + + await item.onSuccess(tx); + } catch (err: any) { + logger.error(`[BoundedTransactionQueue] Processing failed for signature ${item.signature}`, err); + if (this.metricsProvider) { + this.metricsProvider.recordRpcFetchError(item.walletAddress); + } + await item.onFailure(err); + } finally { + this.activeCount--; + logger.debug(`[BoundedTransactionQueue] Finished item ${item.signature}. Active: ${this.activeCount}`); + this.processNext(); + } + } +} \ No newline at end of file diff --git a/src/solana/parser.ts b/src/solana/parser.ts new file mode 100644 index 0000000..33c21d6 --- /dev/null +++ b/src/solana/parser.ts @@ -0,0 +1,140 @@ +import { VersionedTransactionResponse } from '@solana/web3.js'; + +export interface WalletBalanceChanges { + solChange: bigint; + tokenChanges: { mint: string; change: bigint }[]; +} + +export function getAccountKeysArray(tx: VersionedTransactionResponse): any[] { + const message = tx.transaction.message; + if (!message) { + throw new Error('Transaction message is missing or unresolvable'); + } + + // Extract static account keys + let staticKeys: any[] = []; + if (typeof (message as any).getAccountKeys === 'function') { + try { + const keys = (message as any).getAccountKeys(); + if (keys && Array.isArray(keys.staticAccountKeys)) { + staticKeys = keys.staticAccountKeys; + } else if (Array.isArray(keys)) { + staticKeys = keys; + } + } catch { + // fallback to raw fields + } + } + + if (staticKeys.length === 0) { + if (Array.isArray((message as any).staticAccountKeys)) { + staticKeys = (message as any).staticAccountKeys; + } else if (Array.isArray((message as any).accountKeys)) { + staticKeys = (message as any).accountKeys; + } else { + throw new Error('Failed to resolve static account keys from transaction message'); + } + } + + // Extract loaded addresses from meta + let loadedWritable: any[] = []; + let loadedReadonly: any[] = []; + + if (tx.meta?.loadedAddresses) { + if (Array.isArray(tx.meta.loadedAddresses.writable)) { + loadedWritable = tx.meta.loadedAddresses.writable; + } + if (Array.isArray(tx.meta.loadedAddresses.readonly)) { + loadedReadonly = tx.meta.loadedAddresses.readonly; + } + } + + return [...staticKeys, ...loadedWritable, ...loadedReadonly]; +} + +export function calculateBalanceChanges( + tx: VersionedTransactionResponse, + walletAddress: string, +): WalletBalanceChanges { + const accountKeys = getAccountKeysArray(tx); + let walletIndex = -1; + + for (let i = 0; i < accountKeys.length; i++) { + const addr = accountKeys[i]; + const addrStr = addr && typeof addr.toBase58 === 'function' ? addr.toBase58() : String(addr); + if (addrStr === walletAddress) { + walletIndex = i; + break; + } + } + + let solChange = 0n; + if (walletIndex !== -1 && tx.meta) { + const preBalances = tx.meta.preBalances || []; + const postBalances = tx.meta.postBalances || []; + const preVal = preBalances[walletIndex]; + const postVal = postBalances[walletIndex]; + const pre = preVal !== undefined && preVal !== null ? BigInt(preVal) : 0n; + const post = postVal !== undefined && postVal !== null ? BigInt(postVal) : 0n; + solChange = post - pre; + } + + const tokenChanges: { mint: string; change: bigint }[] = []; + if (tx.meta) { + const preTokenBalances = tx.meta.preTokenBalances || []; + const postTokenBalances = tx.meta.postTokenBalances || []; + + const preMap = new Map(); + for (const item of preTokenBalances) { + const owner = item.owner || (walletIndex !== -1 && item.accountIndex === walletIndex ? walletAddress : ''); + if (owner === walletAddress && item.mint && item.uiTokenAmount.amount) { + const val = BigInt(item.uiTokenAmount.amount); + preMap.set(item.mint, (preMap.get(item.mint) || 0n) + val); + } + } + + const postMap = new Map(); + for (const item of postTokenBalances) { + const owner = item.owner || (walletIndex !== -1 && item.accountIndex === walletIndex ? walletAddress : ''); + if (owner === walletAddress && item.mint && item.uiTokenAmount.amount) { + const val = BigInt(item.uiTokenAmount.amount); + postMap.set(item.mint, (postMap.get(item.mint) || 0n) + val); + } + } + + const allMints = new Set([...preMap.keys(), ...postMap.keys()]); + for (const mint of allMints) { + const preAmt = preMap.get(mint) || 0n; + const postAmt = postMap.get(mint) || 0n; + const diff = postAmt - preAmt; + if (diff !== 0n) { + tokenChanges.push({ mint, change: diff }); + } + } + } + + return { + solChange, + tokenChanges, + }; +} + +export function isTransactionRelevant( + tx: VersionedTransactionResponse, + walletAddress: string, +): boolean { + const { solChange, tokenChanges } = calculateBalanceChanges(tx, walletAddress); + + // Sol change is meaningful if it is not just fee (or 0) + // Fees are typically very small, e.g., < 0.005 SOL (5,000,000 lamports) + // Let's check if the native SOL absolute balance change is greater than a standard transaction fee + // 50,000 lamports is the absolute minimum, let's use 10,000,000 lamports (0.01 SOL) to be robustly conservative, + // or simple rule: if absolute solChange is greater than 1,000,000 lamports (0.001 SOL), it's a real transfer/swap. + const absSolChange = solChange < 0n ? -solChange : solChange; + const meaningfulSol = absSolChange > 1_000_000n; // > 0.001 SOL change + + // Token change is meaningful if any token change absolute value > 0 + const meaningfulToken = tokenChanges.length > 0; + + return meaningfulSol || meaningfulToken; +} diff --git a/tests/realMode.test.ts b/tests/realMode.test.ts index 3053580..cabf16f 100644 --- a/tests/realMode.test.ts +++ b/tests/realMode.test.ts @@ -36,6 +36,23 @@ vi.mock('@solana/web3.js', () => { }; }); +// Mock database interactions to avoid live connections +vi.mock('../src/persistence/db.js', () => { + return { + initDb: vi.fn(), + verifyDbHealth: vi.fn().mockResolvedValue(undefined), + closeDb: vi.fn().mockResolvedValue(undefined), + getDbPool: vi.fn().mockImplementation(() => { + return { + query: vi.fn().mockResolvedValue({ + rows: [{ id: 'mock-id', address: '3yF9asA9B7G3Y1G7as78gHJKa7A', name: 'Mock Wallet', is_active: true, created_at: new Date(), updated_at: new Date() }], + rowCount: 1, + }), + }; + }), + }; +}); + describe('Real Solana Observation Mode Unit Tests', () => { let connection: any; @@ -291,4 +308,626 @@ describe('Real Solana Observation Mode Unit Tests', () => { expect(events.map((e) => e.eventType)).toContain('raw_transaction_fetch_failed'); }); }); + + describe('BoundedTransactionQueue Unit Tests', () => { + let mockMetrics: any; + + beforeEach(() => { + mockMetrics = { + recordQueueDepth: vi.fn(), + recordQueueRetries: vi.fn(), + recordQueue429s: vi.fn(), + recordQuoteLatency: vi.fn(), + recordSignalProcessingLatency: vi.fn(), + recordWalletEventReceived: vi.fn(), + recordWalletQueueDepth: vi.fn(), + recordOldestQueuedAge: vi.fn(), + recordDroppedOldestCount: vi.fn(), + recordTransactionRelevance: vi.fn(), + recordRpcRequestRate: vi.fn(), + recordQueueProcessed: vi.fn(), + recordRpcFetchError: vi.fn(), + }; + }); + + it('should respect max concurrency limit and rate limiting delay', async () => { + const { BoundedTransactionQueue } = await import('../src/solana/fetcher.js'); + const mockTx = { slot: 100, blockTime: 123456 }; + connection.getTransaction.mockResolvedValue(mockTx); + + const queue = new BoundedTransactionQueue( + connection as any, + 2, // maxConcurrency = 2 + 100, // minDelayMs = 100ms + 10, // maxQueueSize = 10 + mockMetrics, + ); + queue.start(); + + const successSigs: string[] = []; + const pushItem = (sig: string) => { + return new Promise((resolve) => { + queue.push({ + signature: sig, + walletAddress: 'wallet1', + slot: 100, + detectedAt: new Date(), + correlationId: 'corr', + onSuccess: async () => { + successSigs.push(sig); + resolve(); + }, + onFailure: async () => { + resolve(); + }, + }); + }); + }; + + // Push 3 items concurrently + const p1 = pushItem('sig1'); + const p2 = pushItem('sig2'); + const p3 = pushItem('sig3'); + + // Check that at most 2 are active initially + expect(queue.getActiveCount()).toBeLessThanOrEqual(2); + + await Promise.all([p1, p2, p3]); + + expect(successSigs).toContain('sig1'); + expect(successSigs).toContain('sig2'); + expect(successSigs).toContain('sig3'); + expect(queue.getDepth()).toBe(0); + expect(queue.getActiveCount()).toBe(0); + + queue.stop(); + }); + + it('should drop incoming items and apply backpressure when queue size is exceeded', async () => { + const { BoundedTransactionQueue } = await import('../src/solana/fetcher.js'); + const mockTx = { slot: 100, blockTime: 123456 }; + connection.getTransaction.mockResolvedValue(mockTx); + + const queue = new BoundedTransactionQueue( + connection as any, + 1, // maxConcurrency = 1 + 100, // minDelay = 100 + 2, // maxQueueSize = 2 (allows 2 items to wait in queue) + ); + queue.start(); + + const results: string[] = []; + const errors: string[] = []; + + const pushItem = (sig: string) => { + queue.push({ + signature: sig, + walletAddress: 'wallet1', + slot: 100, + detectedAt: new Date(), + correlationId: 'corr', + onSuccess: async () => { + results.push(sig); + }, + onFailure: async (err) => { + errors.push(`${sig}:${err.message}`); + }, + }); + }; + + // Push 4 items (concurrency=1, max queue=2 -> 1 active, 2 in queue, 4th item triggers drop-oldest of item2) + pushItem('item1'); // goes active immediately (active=1, queue=[]) + pushItem('item2'); // queued (active=1, queue=[item2]) + pushItem('item3'); // queued (active=1, queue=[item2, item3]) + pushItem('item4'); // queue full! item2 is dropped, queue becomes [item3, item4] + + expect(errors).toContain('item2:Dropped from queue due to backpressure'); + + queue.stop(); + }); + + it('should implement HTTP 429 exponential backoff with jitter and successfully retry', async () => { + const mockTx = { slot: 100, blockTime: 123456 }; + // First attempt triggers HTTP 429 rate limit error + const rateLimitErr = { status: 429, message: 'Too Many Requests' }; + connection.getTransaction + .mockRejectedValueOnce(rateLimitErr) + .mockResolvedValueOnce(mockTx); + + const result = await fetchTransactionWithRetry(connection as any, 'sig-429', 3, 10, 50, mockMetrics); + expect(result).toEqual(mockTx); + expect(connection.getTransaction).toHaveBeenCalledTimes(2); + expect(mockMetrics.recordQueue429s).toHaveBeenCalledWith('sig-429'); + }); + + it('should gracefully drain or terminate the queue when stopped during graceful shutdown', async () => { + const { BoundedTransactionQueue } = await import('../src/solana/fetcher.js'); + const queue = new BoundedTransactionQueue( + connection as any, + 1, + 100, + 10, + ); + queue.start(); + + let failed = false; + queue.push({ + signature: 'sig-drain', + walletAddress: 'wallet', + slot: 100, + detectedAt: new Date(), + correlationId: 'corr', + onSuccess: async () => {}, + onFailure: async () => { + failed = true; + }, + }); + + expect(queue.getDepth()).toBe(0); // goes to active immediately + queue.stop(); + + // Pushing after stop must fail immediately + const pushedAfter = queue.push({ + signature: 'sig-after', + walletAddress: 'wallet', + slot: 100, + detectedAt: new Date(), + correlationId: 'corr', + onSuccess: async () => {}, + onFailure: async () => { + failed = true; + }, + }); + + expect(pushedAfter).toBe(false); + expect(failed).toBe(true); + }); + + it('should correctly schedule high-volume wallet without starving another wallet (fair round robin)', async () => { + const { BoundedTransactionQueue } = await import('../src/solana/fetcher.js'); + const mockTx = { slot: 100, blockTime: 123456 }; + connection.getTransaction.mockResolvedValue(mockTx); + + const queue = new BoundedTransactionQueue( + connection as any, + 1, // maxConcurrency = 1 + 10, // minDelay = 10ms + 10, // max size per-wallet + ); + queue.start(); + + const successOrder: string[] = []; + + const pushItem = (sig: string, wallet: string) => { + queue.push({ + signature: sig, + walletAddress: wallet, + slot: 100, + detectedAt: new Date(), + correlationId: 'corr', + onSuccess: async () => { + successOrder.push(sig); + }, + onFailure: async () => {}, + }); + }; + + // Push A1. It executes immediately. + pushItem('A1', 'walletA'); // activeCount=1, walletOrder=[] + + // Now enqueue additional tasks. + // A2 is pushed. It goes to 'walletA' queue. walletOrder becomes ['walletA'] + pushItem('A2', 'walletA'); + // A3 is pushed. It goes to 'walletA' queue. Since 'walletA' is already in walletOrder, order remains ['walletA'] + pushItem('A3', 'walletA'); + // B1 is pushed. It goes to 'walletB' queue. walletOrder becomes ['walletA', 'walletB'] + pushItem('B1', 'walletB'); + + // Now A1 finishes. processNext() is called. + // It retrieves from scheduling order: 'walletOrder.shift()' -> yields 'walletA'! + // So 'A2' is dequeued and executed next. + // Since 'walletA' still has 'A3', it is re-queued at the back of the order. + // walletOrder becomes: ['walletB', 'walletA'] + // + // Then A2 finishes. processNext() is called. + // It shifts 'walletOrder' -> yields 'walletB'! + // So 'B1' is dequeued and executed third. + // + // Finally B1 finishes. processNext() shifts -> yields 'walletA' -> executes 'A3'. + // + // Hence successOrder will be: ['A1', 'A2', 'B1', 'A3'] + // This proves B1 is scheduled fairly and is NOT starved by A3 even though A was queued first! + + // Process and wait for all tasks to settle + await new Promise((resolve) => setTimeout(resolve, 150)); + + expect(successOrder[0]).toBe('A1'); + expect(successOrder[1]).toBe('A2'); + expect(successOrder[2]).toBe('B1'); // Fair scheduling ensures B1 executes third, before A3! + expect(successOrder[3]).toBe('A3'); + + queue.stop(); + }); + + it('should drop oldest items and preserve newest events under backpressure', async () => { + const { BoundedTransactionQueue } = await import('../src/solana/fetcher.js'); + const mockTx = { slot: 100, blockTime: 123456 }; + connection.getTransaction.mockResolvedValue(mockTx); + + const queue = new BoundedTransactionQueue( + connection as any, + 1, // maxConcurrency = 1 + 50, // minDelay = 50ms + 2, // max queue size per-wallet = 2 + ); + queue.start(); + + const results: string[] = []; + const errors: string[] = []; + + const pushItem = (sig: string) => { + queue.push({ + signature: sig, + walletAddress: 'wallet1', + slot: 100, + detectedAt: new Date(), + correlationId: 'corr', + onSuccess: async () => { + results.push(sig); + }, + onFailure: async (err) => { + errors.push(`${sig}:${err.message}`); + }, + }); + }; + + pushItem('T1'); // Runs immediately + pushItem('T2'); // Queued (pos 0) + pushItem('T3'); // Queued (pos 1) + pushItem('T4'); // Queued full! Drops oldest ('T2'), appends T4. Queue is now: [T3, T4] + + expect(errors).toContain('T2:Dropped from queue due to backpressure'); + + await new Promise((resolve) => setTimeout(resolve, 200)); + + expect(results).toContain('T1'); + expect(results).not.toContain('T2'); // dropped + expect(results).toContain('T3'); + expect(results).toContain('T4'); // newest preserved! + + queue.stop(); + }); + + it('should prove backpressure drop is not reported as fetch failure, generates no stack trace, and metrics distinguish it', async () => { + const { BoundedTransactionQueue, BackpressureDropError } = await import('../src/solana/fetcher.js'); + const { logger } = await import('../src/logging/index.js'); + const loggerWarnSpy = vi.spyOn(logger, 'warn'); + const loggerErrorSpy = vi.spyOn(logger, 'error'); + + const queue = new BoundedTransactionQueue( + connection as any, + 1, // maxConcurrency = 1 + 10, // minDelay = 10ms + 1, // max queue size per-wallet = 1 + mockMetrics, + ); + queue.start(); + + let capturedError: any = null; + + queue.push({ + signature: 'T1', + walletAddress: 'wallet1', + slot: 100, + detectedAt: new Date(), + correlationId: 'corr1', + onSuccess: async () => {}, + onFailure: async () => {}, + }); + + queue.push({ + signature: 'T2', + walletAddress: 'wallet1', + slot: 100, + detectedAt: new Date(), + correlationId: 'corr2', + onSuccess: async () => {}, + onFailure: async (err) => { + capturedError = err; + }, + }); + + // T3 will trigger drop of T2 + queue.push({ + signature: 'T3', + walletAddress: 'wallet1', + slot: 100, + detectedAt: new Date(), + correlationId: 'corr3', + onSuccess: async () => {}, + onFailure: async () => {}, + }); + + expect(capturedError).toBeInstanceOf(BackpressureDropError); + expect(capturedError.stack).toBeUndefined(); + + // Verify that backpressure drop logs WARN, not ERROR + const backpressureWarns = loggerWarnSpy.mock.calls.filter(call => + call[0] && call[0].includes('BACKPRESSURE') + ); + expect(backpressureWarns.length).toBeGreaterThan(0); + + const backpressureErrors = loggerErrorSpy.mock.calls.filter(call => + call[0] && call[0].includes('BACKPRESSURE') + ); + expect(backpressureErrors.length).toBe(0); + + // Verify operational metrics: recordDroppedOldestCount called, recordRpcFetchError NOT called + expect(mockMetrics.recordDroppedOldestCount).toHaveBeenCalledWith('wallet1'); + expect(mockMetrics.recordRpcFetchError).not.toHaveBeenCalled(); + + queue.stop(); + loggerWarnSpy.mockRestore(); + loggerErrorSpy.mockRestore(); + }); + + it('should prove one rate-limit wait produces only one scheduling/log event and does not spam logs', async () => { + const { BoundedTransactionQueue } = await import('../src/solana/fetcher.js'); + const { logger } = await import('../src/logging/index.js'); + const loggerDebugSpy = vi.spyOn(logger, 'debug'); + + const queue = new BoundedTransactionQueue( + connection as any, + 1, // maxConcurrency = 1 + 1000, // minDelay = 1000ms (so we trigger rate limit) + 10, // max queue size per-wallet = 10 + mockMetrics, + ); + queue.start(); + + // Push item 1 (executed immediately, updates lastRequestTime) + queue.push({ + signature: 'R1', + walletAddress: 'wallet1', + slot: 100, + detectedAt: new Date(), + correlationId: 'corr1', + onSuccess: async () => {}, + onFailure: async () => {}, + }); + + // Push 5 more items concurrently (within rate-limit interval) + for (let i = 2; i <= 6; i++) { + queue.push({ + signature: `R${i}`, + walletAddress: 'wallet1', + slot: 100, + detectedAt: new Date(), + correlationId: `corr${i}`, + onSuccess: async () => {}, + onFailure: async () => {}, + }); + } + + // Wait for R1 to finish and trigger the rate limit wait + await new Promise((resolve) => setTimeout(resolve, 50)); + + // Check the debug logs for "Global rate limit: waiting" + const rateLimitDebugs = loggerDebugSpy.mock.calls.filter(call => + call[0] && call[0].includes('Global rate limit: waiting') + ); + + // Even with 5 pushes under rate limit, we should only have ONE rate limit waiting log event + expect(rateLimitDebugs.length).toBe(1); + + queue.stop(); + loggerDebugSpy.mockRestore(); + }); + + it('should classify fetched transactions correctly based on SOL/SPL token balance changes', async () => { + const { isTransactionRelevant } = await import('../src/solana/parser.js'); + + // 1. Irrelevant payload (no balance changes or only extremely tiny SOL shift simulating fee) + const irrelevantTx: any = { + transaction: { + message: { + getAccountKeys: () => ({ + staticAccountKeys: [{ toBase58: () => 'my-wallet' }], + }), + }, + }, + meta: { + preBalances: [1000000000n], + postBalances: [1000000000n], // exactly 0 change + preTokenBalances: [], + postTokenBalances: [], + }, + }; + + // 2. Meaningful SOL transfer payload + const relevantSolTx: any = { + transaction: { + message: { + getAccountKeys: () => ({ + staticAccountKeys: [{ toBase58: () => 'my-wallet' }], + }), + }, + }, + meta: { + preBalances: [1000000000], + postBalances: [900000000], // -100,000,000 lamports (-0.1 SOL) change + preTokenBalances: [], + postTokenBalances: [], + }, + }; + + // 3. Meaningful SPL Token swap payload + const relevantTokenTx: any = { + transaction: { + message: { + getAccountKeys: () => ({ + staticAccountKeys: [{ toBase58: () => 'my-wallet' }], + }), + }, + }, + meta: { + preBalances: [1000000000], + postBalances: [1000000000], // 0 SOL change + preTokenBalances: [ + { owner: 'my-wallet', mint: 'token-mint-1', uiTokenAmount: { amount: '1000' } }, + ], + postTokenBalances: [ + { owner: 'my-wallet', mint: 'token-mint-1', uiTokenAmount: { amount: '5000' } }, // +4000 change + ], + }, + }; + + expect(isTransactionRelevant(irrelevantTx, 'my-wallet')).toBe(false); + expect(isTransactionRelevant(relevantSolTx, 'my-wallet')).toBe(true); + expect(isTransactionRelevant(relevantTokenTx, 'my-wallet')).toBe(true); + }); + + it('should correctly resolve accounts and relevance for legacy transactions', async () => { + const { isTransactionRelevant } = await import('../src/solana/parser.js'); + + const legacyTx: any = { + transaction: { + message: { + accountKeys: [{ toBase58: () => 'my-wallet' }, { toBase58: () => 'other-acct' }], + }, + }, + meta: { + preBalances: [2000000000, 50000000], + postBalances: [1000000000, 105000000], // SOL shift: -1 SOL for my-wallet (relevant) + preTokenBalances: [], + postTokenBalances: [], + }, + }; + + expect(isTransactionRelevant(legacyTx, 'my-wallet')).toBe(true); + expect(isTransactionRelevant(legacyTx, 'other-acct')).toBe(true); + }); + + it('should correctly resolve accounts and relevance for v0 transactions without address lookup tables', async () => { + const { isTransactionRelevant } = await import('../src/solana/parser.js'); + + const v0TxWithoutAlts: any = { + transaction: { + message: { + staticAccountKeys: [{ toBase58: () => 'my-wallet' }, { toBase58: () => 'other-acct' }], + }, + }, + meta: { + preBalances: [1000000000, 50000000], + postBalances: [1000000000, 50000000], // no SOL shift + preTokenBalances: [ + { owner: 'my-wallet', mint: 'token-mint-v0', uiTokenAmount: { amount: '10' } } + ], + postTokenBalances: [ + { owner: 'my-wallet', mint: 'token-mint-v0', uiTokenAmount: { amount: '110' } } // relevant SPL shift + ], + }, + }; + + expect(isTransactionRelevant(v0TxWithoutAlts, 'my-wallet')).toBe(true); + }); + + it('should correctly resolve accounts and relevance for v0 transactions with loaded addresses (ALTs)', async () => { + const { isTransactionRelevant, getAccountKeysArray } = await import('../src/solana/parser.js'); + + const v0TxWithAlts: any = { + transaction: { + message: { + staticAccountKeys: [{ toBase58: () => 'static-1' }], + }, + }, + meta: { + loadedAddresses: { + writable: [{ toBase58: () => 'loaded-writable-1' }], + readonly: [{ toBase58: () => 'loaded-readonly-1' }, { toBase58: () => 'loaded-readonly-2' }] + }, + preBalances: [5000000000, 1000000000, 50000000, 50000000], + postBalances: [4900000000, 1100000000, 50000000, 50000000], // transfer 0.1 SOL from static-1 to loaded-writable-1 + preTokenBalances: [], + postTokenBalances: [], + }, + }; + + // Verify full index mapping: + // 0 -> static-1, 1 -> loaded-writable-1, 2 -> loaded-readonly-1, 3 -> loaded-readonly-2 + const keys = getAccountKeysArray(v0TxWithAlts).map(k => k.toBase58()); + expect(keys).toEqual(['static-1', 'loaded-writable-1', 'loaded-readonly-1', 'loaded-readonly-2']); + + // static-1 had post-pre = 4.9B - 5B = -100M lamports (relevant SOL change) + expect(isTransactionRelevant(v0TxWithAlts, 'static-1')).toBe(true); + + // loaded-writable-1 had post-pre = 1.1B - 1B = 100M lamports (relevant SOL change) + expect(isTransactionRelevant(v0TxWithAlts, 'loaded-writable-1')).toBe(true); + + // loaded-readonly-1 had post-pre = 50M - 50M = 0 lamports (irrelevant) + expect(isTransactionRelevant(v0TxWithAlts, 'loaded-readonly-1')).toBe(false); + }); + + it('should throw an error on account resolution or parsing failures as a distinct error state', async () => { + const { isTransactionRelevant } = await import('../src/solana/parser.js'); + + const malformedTx: any = { + transaction: { + // message is missing + }, + }; + + await expect(async () => { + isTransactionRelevant(malformedTx, 'any-wallet'); + }).rejects.toThrow('Transaction message is missing or unresolvable'); + }); + }); + + describe('Application Real Mode vs Mock Mode Loop Behavior', () => { + it('should NOT start SimulationScheduler processing loop in real mode and keep pending simulated executions untouched', async () => { + const oldMode = process.env.WALLET_WATCHER_MODE; + process.env.WALLET_WATCHER_MODE = 'real'; + process.env.WATCHED_WALLETS = '3yF9asA9B7G3Y1G7as78gHJKa7A'; + + try { + const { Application } = await import('../src/application/index.js'); + const app = new Application(); + await app.bootstrap(); + + // Proves that no interval id was created/started for the simulated loop + expect(app.simulationIntervalId).toBeNull(); + + await app.shutdown('SIGINT'); + } finally { + if (oldMode) { + process.env.WALLET_WATCHER_MODE = oldMode; + } else { + delete process.env.WALLET_WATCHER_MODE; + } + } + }); + + it('should start SimulationScheduler processing loop in mock mode', async () => { + const oldMode = process.env.WALLET_WATCHER_MODE; + process.env.WALLET_WATCHER_MODE = 'mock'; + process.env.WATCHED_WALLETS = '3yF9asA9B7G3Y1G7as78gHJKa7A'; + + try { + const { Application } = await import('../src/application/index.js'); + const app = new Application(); + await app.bootstrap(); + + // Proves that interval id WAS started for the simulated loop + expect(app.simulationIntervalId).not.toBeNull(); + + await app.shutdown('SIGINT'); + } finally { + if (oldMode) { + process.env.WALLET_WATCHER_MODE = oldMode; + } else { + delete process.env.WALLET_WATCHER_MODE; + } + } + }); + }); });