quick/memecoin-botV2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity Open in Code-Server

test1 successful

Browse Source
This commit is contained in:
quick
2026-08-16 12:21:27 +00:00
parent 37ee805b96
commit a85647572a
7 changed files with 1273 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files

118
src/application/index.ts
View File

@@ -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<void> {
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,38 +186,50 @@ 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');
const queued = this.transactionQueue.push({
signature: observation.signature,
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;
@@ -224,9 +253,6 @@ export class Application {
await rawTxRepo.saveTransaction(rawTx);
const totalDuration = Date.now() - detectedAt.getTime();
this.metricsProvider?.recordSignalProcessingLatency(totalDuration);
await journalService.log(correlationId, 'raw_transaction_persisted', {
signature: observation.signature,
slot: observation.slot,
@@ -236,6 +262,35 @@ export class Application {
} 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,
});
}
},
});
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,
error: 'Queue backpressure: item dropped',
});
}
// STOP (No strategy evaluation, scheduler or copy signal creation in real mode!)
return;
@@ -310,7 +365,8 @@ export class Application {
// 7. Start on-chain watcher
await this.watcher.startWatching();
// 8. Run simulator poller non-blockingly every 500ms
// 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());
@@ -318,6 +374,7 @@ export class Application {
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();
}

16
src/config/index.ts
View File

@@ -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(''),
});

13
src/domain/interfaces.ts
View File

@@ -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;
}

66
src/metrics/index.ts
View File

@@ -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<void> {
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 {}
}

271
src/solana/fetcher.ts
View File

@@ -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<VersionedTransactionResponse> {
let attempt = 0;
let delay = initialDelayMs;
@@ -24,15 +26,38 @@ export async function fetchTransactionWithRetry(
}
logger.warn(`[TransactionFetcher] Transaction ${signature} not found on attempt ${attempt + 1}. Retrying...`);
} catch (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++;
if (attempt >= maxRetries) {
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<void>;
onFailure: (error: Error) => Promise<void>;
}
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<string, QueueItem[]>();
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<void> {
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<void> {
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();
}
}
}

140
src/solana/parser.ts Normal file
View File

@@ -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<string, bigint>();
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<string, bigint>();
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;
}

639
tests/realMode.test.ts
View File

@@ -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<void>((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;
}
}
});
});
});