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

fixing empty transaction

Browse Source
This commit is contained in:
quick
2026-08-16 19:44:30 +00:00
parent a85647572a
commit 9e23b248e5
7 changed files with 357 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
src/application/index.ts
View File

@@ -158,6 +158,15 @@ export class Application {
return;
}
// Lightweight pre-fetch filter using websocket/log data
if (wsPayload.logs && Array.isArray(wsPayload.logs)) {
const { isLogIrrelevant } = await import('../solana/parser.js');
if (isLogIrrelevant(wsPayload.logs)) {
logger.info(`[RealMode] Pre-filtered obviously irrelevant transaction signature ${observation.signature} based on log analysis.`, { correlationId });
return;
}
}
logger.info(
`[RealMode] Activity detected for wallet ${observation.walletAddress}. Signature: ${observation.signature}, Slot: ${observation.slot}`,
{ correlationId },
@@ -188,7 +197,9 @@ export class Application {
// Push item onto the bounded queue for non-blocking execution
if (!this.transactionQueue) {
logger.error('[RealMode] Transaction queue not initialized!');
if (!this.isShuttingDown) {
logger.error('[RealMode] Transaction queue not initialized!');
}
return;
}
@@ -385,20 +396,23 @@ export class Application {
logger.warn(`Shutdown signal received (${signal}). Commencing graceful shutdown...`);
// 1. Stop accepting new watcher events / unsubscribe first
if (this.watcher) {
await this.watcher.stopWatching();
}
// 2. Clear intervals and allow pending callbacks to settle cleanly
if (this.simulationIntervalId) {
clearInterval(this.simulationIntervalId);
this.simulationIntervalId = null;
}
// 3. Stop and clear transaction queues
if (this.transactionQueue) {
this.transactionQueue.stop();
this.transactionQueue = null;
}
if (this.watcher) {
await this.watcher.stopWatching();
}
try {
const config = loadConfig();
const journalRepo = new PgEventJournalRepository();
@@ -413,6 +427,7 @@ export class Application {
logger.error('Could not log application_stopping event during shutdown', err);
}
// 4. Close InfluxDB / PostgreSQL metrics & data connections
if (this.metricsProvider && 'close' in this.metricsProvider) {
await this.metricsProvider.close();
}

3
src/domain/interfaces.ts
View File

@@ -143,4 +143,7 @@ export interface IMetricsProvider {
recordRpcRequestRate(requestsPerSecond: number): void;
recordQueueProcessed(wallet: string): void;
recordRpcFetchError(wallet: string): void;
recordWalletState(wallet: string, isHot: boolean): void;
recordWalletIncomingRate(wallet: string, rate: number): void;
recordWalletProcessedRate(wallet: string, rate: number): void;
}

18
src/metrics/index.ts
View File

@@ -146,6 +146,21 @@ export class InfluxMetricsProvider implements IMetricsProvider {
this.writePoint(p);
}
recordWalletState(wallet: string, isHot: boolean): void {
const p = new Point('wallet_state').tag('wallet', wallet).intField('is_hot', isHot ? 1 : 0);
this.writePoint(p);
}
recordWalletIncomingRate(wallet: string, rate: number): void {
const p = new Point('wallet_incoming_rate').tag('wallet', wallet).floatField('rate', rate);
this.writePoint(p);
}
recordWalletProcessedRate(wallet: string, rate: number): void {
const p = new Point('wallet_processed_rate').tag('wallet', wallet).floatField('rate', rate);
this.writePoint(p);
}
async close(): Promise<void> {
if (this.writeApi) {
logger.info('Closing InfluxDB metrics writer...');
@@ -180,4 +195,7 @@ export class MockMetricsProvider implements IMetricsProvider {
recordRpcRequestRate(_requestsPerSecond: number): void {}
recordQueueProcessed(_wallet: string): void {}
recordRpcFetchError(_wallet: string): void {}
recordWalletState(_wallet: string, _isHot: boolean): void {}
recordWalletIncomingRate(_wallet: string, _rate: number): void {}
recordWalletProcessedRate(_wallet: string, _rate: number): void {}
}

137
src/solana/fetcher.ts
View File

@@ -9,17 +9,20 @@ export async function fetchTransactionWithRetry(
initialDelayMs = 500,
maxDelayMs = 8000,
metricsProvider?: IMetricsProvider,
transactionDetails: 'accounts' | 'full' = 'full',
onRateLimit?: () => void,
): Promise<VersionedTransactionResponse> {
let attempt = 0;
let delay = initialDelayMs;
while (attempt < maxRetries) {
try {
logger.debug(`[TransactionFetcher] Fetching transaction ${signature} (attempt ${attempt + 1}/${maxRetries})...`);
logger.debug(`[TransactionFetcher] Fetching transaction ${signature} (attempt ${attempt + 1}/${maxRetries}, details: ${transactionDetails})...`);
const tx = await connection.getTransaction(signature, {
commitment: 'confirmed',
maxSupportedTransactionVersion: 0,
});
transactionDetails,
} as any);
if (tx) {
return tx;
@@ -33,6 +36,9 @@ export async function fetchTransactionWithRetry(
if (metricsProvider) {
metricsProvider.recordQueue429s(signature);
}
if (onRateLimit) {
onRateLimit();
}
// Use exponential backoff with jitter for HTTP 429
const jitter = Math.random() * 200;
const rateLimitDelay = Math.min(delay * 3 + jitter, maxDelayMs * 2);
@@ -103,13 +109,65 @@ export class BoundedTransactionQueue {
private rateSecondTimer: NodeJS.Timeout | null = null;
private rateLimitTimeout: NodeJS.Timeout | null = null;
// Wallet tracking stats for rate and state tracking
private walletIncomingCountLastSecond = new Map<string, number>();
private walletProcessedCountLastSecond = new Map<string, number>();
private walletIncomingRate = new Map<string, number>();
private walletProcessedRate = new Map<string, number>();
private walletStates = new Map<string, 'NORMAL' | 'HOT'>();
// Adaptive rate-limiting variables
private adaptiveDelayMs: number;
constructor(
private connection: Connection,
private maxConcurrency: number,
private minDelayMs: number,
private maxQueueSize: number, // Bounded limit PER wallet queue
private metricsProvider?: IMetricsProvider,
) {}
) {
this.adaptiveDelayMs = this.minDelayMs;
}
public onRateLimitHit(): void {
const oldDelay = this.adaptiveDelayMs;
this.adaptiveDelayMs = Math.min(this.adaptiveDelayMs * 1.5, 5000);
logger.warn(`[BoundedTransactionQueue] Rate limit (429) detected. Increasing adaptive minDelayMs from ${oldDelay}ms to ${this.adaptiveDelayMs}ms`);
}
public onFetchSuccess(): void {
if (this.adaptiveDelayMs > this.minDelayMs) {
const oldDelay = this.adaptiveDelayMs;
this.adaptiveDelayMs = Math.max(this.minDelayMs, this.adaptiveDelayMs - 50);
logger.debug(`[BoundedTransactionQueue] Successful fetch. Decaying adaptive minDelayMs from ${oldDelay}ms to ${this.adaptiveDelayMs}ms`);
}
}
private updateWalletState(wallet: string): void {
const qSize = this.walletQueues.get(wallet)?.length || 0;
const incRate = this.walletIncomingRate.get(wallet) || 0;
const isNearCapacity = qSize >= this.maxQueueSize * 0.8;
const isHighRate = incRate > 10;
const oldState = this.walletStates.get(wallet) || 'NORMAL';
let newState: 'NORMAL' | 'HOT' = 'NORMAL';
if (isNearCapacity || isHighRate) {
newState = 'HOT';
}
if (oldState !== newState) {
this.walletStates.set(wallet, newState);
logger.warn(`[BoundedTransactionQueue] Wallet ${wallet} transitioned from ${oldState} to ${newState}. Queue depth: ${qSize}, Incoming rate: ${incRate}/s`);
}
if (this.metricsProvider) {
this.metricsProvider.recordWalletState(wallet, newState === 'HOT');
}
}
public getWalletState(wallet: string): 'NORMAL' | 'HOT' {
return this.walletStates.get(wallet) || 'NORMAL';
}
public start(): void {
if (this.running) return;
@@ -121,6 +179,24 @@ export class BoundedTransactionQueue {
this.metricsProvider.recordRpcRequestRate(this.requestCountLastSecond);
}
this.requestCountLastSecond = 0;
for (const wallet of this.walletQueues.keys()) {
const incomingCount = this.walletIncomingCountLastSecond.get(wallet) || 0;
const processedCount = this.walletProcessedCountLastSecond.get(wallet) || 0;
this.walletIncomingRate.set(wallet, incomingCount);
this.walletProcessedRate.set(wallet, processedCount);
this.walletIncomingCountLastSecond.set(wallet, 0);
this.walletProcessedCountLastSecond.set(wallet, 0);
if (this.metricsProvider) {
this.metricsProvider.recordWalletIncomingRate(wallet, incomingCount);
this.metricsProvider.recordWalletProcessedRate(wallet, processedCount);
}
this.updateWalletState(wallet);
}
}, 1000);
}
@@ -161,6 +237,9 @@ export class BoundedTransactionQueue {
return false;
}
const currentIncoming = this.walletIncomingCountLastSecond.get(item.walletAddress) || 0;
this.walletIncomingCountLastSecond.set(item.walletAddress, currentIncoming + 1);
if (this.metricsProvider) {
this.metricsProvider.recordWalletEventReceived(item.walletAddress);
}
@@ -206,6 +285,8 @@ export class BoundedTransactionQueue {
this.walletOrder.push(item.walletAddress);
}
this.updateWalletState(item.walletAddress);
logger.debug(`[BoundedTransactionQueue] Queued signature ${item.signature} for wallet ${item.walletAddress}. Depth: ${wQueue.length}`);
// Trigger process next non-blockingly
@@ -228,10 +309,10 @@ export class BoundedTransactionQueue {
// Enforcement of rate limiting delay
const now = Date.now();
const timeSinceLast = now - this.lastRequestTime;
if (timeSinceLast < this.minDelayMs) {
const waitTime = this.minDelayMs - timeSinceLast;
if (timeSinceLast < this.adaptiveDelayMs) {
const waitTime = this.adaptiveDelayMs - timeSinceLast;
if (!this.rateLimitTimeout) {
logger.debug(`[BoundedTransactionQueue] Global rate limit: waiting ${waitTime}ms`);
logger.debug(`[BoundedTransactionQueue] Global rate limit: waiting ${waitTime}ms (adaptive delay: ${this.adaptiveDelayMs}ms)`);
this.rateLimitTimeout = setTimeout(() => {
this.rateLimitTimeout = null;
this.processNext();
@@ -255,6 +336,8 @@ export class BoundedTransactionQueue {
this.walletOrder.push(walletAddr);
}
this.updateWalletState(walletAddr);
if (this.metricsProvider) {
this.metricsProvider.recordWalletQueueDepth(walletAddr, wQueue.length);
this.metricsProvider.recordQueueDepth(this.getDepth());
@@ -276,15 +359,51 @@ export class BoundedTransactionQueue {
private async executeFetch(item: QueueItem): Promise<void> {
const fetchStartTime = Date.now();
try {
const tx = await fetchTransactionWithRetry(
// 1. Cheap pre-fetch to determine relevance
logger.debug(`[BoundedTransactionQueue] Pre-filtering transaction ${item.signature} with cheap accounts-only RPC call...`);
let tx = await fetchTransactionWithRetry(
this.connection,
item.signature,
5,
500,
4000,
this.metricsProvider,
'accounts',
() => this.onRateLimitHit(),
);
// 2. Classify transaction relevance (with robustness for test mocks lacking full transaction structures)
let relevant = true;
let shouldFetchFull = true;
if (tx && tx.transaction && tx.transaction.message) {
const { isTransactionRelevant } = await import('./parser.js');
relevant = isTransactionRelevant(tx, item.walletAddress);
const { loadConfig } = await import('../config/index.js');
const config = loadConfig();
shouldFetchFull = relevant || config.STORE_IRRELEVANT_RAW_TRANSACTIONS;
}
// 3. If relevant (or storing irrelevant is enabled), fetch full transaction details
if (shouldFetchFull) {
logger.debug(`[BoundedTransactionQueue] Transaction ${item.signature} is RELEVANT or requires full details. Fetching full transaction details...`);
tx = await fetchTransactionWithRetry(
this.connection,
item.signature,
5,
500,
4000,
this.metricsProvider,
'full',
() => this.onRateLimitHit(),
);
} else {
logger.debug(`[BoundedTransactionQueue] Pre-filtered: Transaction ${item.signature} is IRRELEVANT. Skipping full fetch to prevent RPC 429 throttling.`);
}
this.onFetchSuccess();
const fetchEndTime = Date.now();
const fetchDuration = fetchEndTime - fetchStartTime;
@@ -293,6 +412,9 @@ export class BoundedTransactionQueue {
this.metricsProvider.recordQueueProcessed(item.walletAddress);
}
const procCount = this.walletProcessedCountLastSecond.get(item.walletAddress) || 0;
this.walletProcessedCountLastSecond.set(item.walletAddress, procCount + 1);
await item.onSuccess(tx);
} catch (err: any) {
logger.error(`[BoundedTransactionQueue] Processing failed for signature ${item.signature}`, err);
@@ -302,6 +424,7 @@ export class BoundedTransactionQueue {
await item.onFailure(err);
} finally {
this.activeCount--;
this.updateWalletState(item.walletAddress);
logger.debug(`[BoundedTransactionQueue] Finished item ${item.signature}. Active: ${this.activeCount}`);
this.processNext();
}

15
src/solana/parser.ts
View File

@@ -138,3 +138,18 @@ export function isTransactionRelevant(
return meaningfulSol || meaningfulToken;
}
export function isLogIrrelevant(logs: string[]): boolean {
if (!logs || logs.length === 0) return false;
// High confidence irrelevant signature patterns:
// Validator vote transactions (contain Vote111111111111111111111111111111111111111)
const hasVote = logs.some((line) =>
line.includes('Vote111111111111111111111111111111111111111'),
);
if (hasVote) {
return true;
}
return false;
}

129
src/solana/watcher.ts
View File

@@ -5,9 +5,10 @@ import { logger } from '../logging/index.js';
export class RealWalletWatcher implements IWalletWatcher {
private callback: ((obs: WalletTransactionObservation) => Promise<void>) | null = null;
private connection: Connection | null = null;
private subscriptionIds: number[] = [];
public connection: Connection | null = null;
private active = false;
private lastKnownSignatures = new Map<string, string>();
private pollTimeouts: NodeJS.Timeout[] = [];
constructor(
private watchedWallets: string[],
@@ -29,35 +30,95 @@ export class RealWalletWatcher implements IWalletWatcher {
for (const wallet of this.watchedWallets) {
try {
const pubkey = new PublicKey(wallet);
const subId = this.connection.onLogs(
// Get the latest signature on startup so we only watch for NEW transactions
const initialSigs = await this.connection.getSignaturesForAddress(
pubkey,
async (logs, context) => {
if (!this.callback) return;
if (!logs.signature || !this.active) return;
const correlationId = randomUUID();
const observation: WalletTransactionObservation = {
signature: logs.signature,
slot: context.slot,
walletAddress: wallet,
timestamp: new Date(),
correlationId,
};
logger.info(`[RealWalletWatcher] Activity detected for wallet ${wallet}. Signature: ${logs.signature}, Slot: ${context.slot}`, { correlationId });
try {
await this.callback(observation);
} catch (error) {
logger.error(`[RealWalletWatcher] Error in observation handler for signature ${logs.signature}`, error);
}
},
{ limit: 1 },
'confirmed'
);
this.subscriptionIds.push(subId);
logger.info(`[RealWalletWatcher] Subscribed to logs for wallet ${wallet} (subId: ${subId})`);
const firstSig = initialSigs[0];
if (firstSig) {
this.lastKnownSignatures.set(wallet, firstSig.signature);
logger.info(`[RealWalletWatcher] Wallet ${wallet} initial signature checkpoint: ${firstSig.signature}`);
} else {
logger.info(`[RealWalletWatcher] Wallet ${wallet} has no previous transactions.`);
}
// Start adaptive recursive setTimeout polling loop
const pollWallet = async (currentDelay = 2000) => {
if (!this.active) return;
let nextDelay = currentDelay;
try {
const lastSig = this.lastKnownSignatures.get(wallet);
const options: any = { limit: 20 };
if (lastSig) {
options.until = lastSig;
}
const sigInfos = await this.connection!.getSignaturesForAddress(
pubkey,
options,
'confirmed'
);
// On success, slowly decay nextDelay back to default 2000ms
nextDelay = Math.max(2000, currentDelay - 500);
if (sigInfos.length > 0) {
// Update last known signature to the newest one (first item in response)
const newestSig = sigInfos[0];
if (newestSig) {
this.lastKnownSignatures.set(wallet, newestSig.signature);
}
// Process from oldest to newest
const newSigs = [...sigInfos].reverse();
for (const sigInfo of newSigs) {
if (!this.callback || !this.active) return;
const correlationId = randomUUID();
const observation: WalletTransactionObservation & { rawPayload?: any } = {
signature: sigInfo.signature,
slot: sigInfo.slot,
walletAddress: wallet,
timestamp: new Date(),
correlationId,
rawPayload: sigInfo,
};
logger.info(`[RealWalletWatcher] Activity detected for wallet ${wallet}. Signature: ${sigInfo.signature}, Slot: ${sigInfo.slot}`, { correlationId });
try {
await this.callback(observation);
} catch (error) {
logger.error(`[RealWalletWatcher] Error in observation handler for signature ${sigInfo.signature}`, error);
}
}
}
} catch (err: any) {
const isRateLimit = err && (err.status === 429 || String(err.message).includes('429') || String(err).includes('429'));
if (isRateLimit) {
nextDelay = Math.min(10000, currentDelay * 2);
logger.warn(`[RealWalletWatcher] HTTP 429 Rate Limit encountered during signatures poll for wallet ${wallet}. Backing off polling delay to ${nextDelay}ms.`);
} else {
logger.error(`[RealWalletWatcher] Error polling transactions for wallet ${wallet}`, err);
}
} finally {
if (this.active) {
const timeoutId = setTimeout(() => pollWallet(nextDelay), nextDelay);
this.pollTimeouts.push(timeoutId);
}
}
};
// Start the first poll
const timeoutId = setTimeout(() => pollWallet(2000), 2000);
this.pollTimeouts.push(timeoutId);
logger.info(`[RealWalletWatcher] Polling transaction signatures for wallet ${wallet} every 2000ms`);
} catch (err) {
logger.error(`[RealWalletWatcher] Failed to subscribe to logs for wallet ${wallet}`, err);
logger.error(`[RealWalletWatcher] Failed to initialize polling for wallet ${wallet}`, err);
}
}
} catch (err) {
@@ -69,17 +130,11 @@ export class RealWalletWatcher implements IWalletWatcher {
async stopWatching(): Promise<void> {
this.active = false;
logger.info('Stopping RealWalletWatcher...');
if (this.connection) {
for (const subId of this.subscriptionIds) {
try {
await this.connection.removeOnLogsListener(subId);
} catch (err) {
logger.error(`[RealWalletWatcher] Error removing onLogs listener ${subId}`, err);
}
}
this.subscriptionIds = [];
this.connection = null;
for (const timeoutId of this.pollTimeouts) {
clearTimeout(timeoutId);
}
this.pollTimeouts = [];
this.connection = null;
logger.info('Stopped RealWalletWatcher.');
}

93
tests/realMode.test.ts
View File

@@ -15,12 +15,14 @@ import { randomUUID } from 'crypto';
const mockOnLogs = vi.fn().mockReturnValue(12345);
const mockRemoveOnLogsListener = vi.fn().mockResolvedValue(true);
const mockGetTransaction = vi.fn();
const mockGetSignaturesForAddress = vi.fn().mockResolvedValue([]);
vi.mock('@solana/web3.js', () => {
class MockConnection {
onLogs = mockOnLogs;
removeOnLogsListener = mockRemoveOnLogsListener;
getTransaction = mockGetTransaction;
getSignaturesForAddress = mockGetSignaturesForAddress;
}
class MockPublicKey {
@@ -96,26 +98,17 @@ describe('Real Solana Observation Mode Unit Tests', () => {
// 2. RealWalletWatcher Subscription and Shutdown Behavior
describe('RealWalletWatcher Lifecycle & Reconnect Behavior', () => {
it('should subscribe to logs for all watched wallets on startup', async () => {
it('should query signatures on startup to set checkpoint and poll periodically', async () => {
const wallets = ['3yF9asA9B7G3Y1G7as78gHJKa7A', '4yF9asA9B7G3Y1G7as78gHJKa7B'];
mockGetSignaturesForAddress.mockResolvedValue([
{ signature: 'sig1', slot: 100, err: null }
]);
const watcher = new RealWalletWatcher(wallets, 'http://dummy-rpc', 'wss://dummy-wss');
await watcher.startWatching();
expect(connection.onLogs).toHaveBeenCalledTimes(2);
expect(connection.onLogs).toHaveBeenNthCalledWith(1, expect.any(Object), expect.any(Function), 'confirmed');
expect(connection.onLogs).toHaveBeenNthCalledWith(2, expect.any(Object), expect.any(Function), 'confirmed');
});
it('should unsubscribe from all logs listeners on shutdown', async () => {
const wallets = ['3yF9asA9B7G3Y1G7as78gHJKa7A'];
const watcher = new RealWalletWatcher(wallets, 'http://dummy-rpc', 'wss://dummy-wss');
await watcher.startWatching();
expect(mockGetSignaturesForAddress).toHaveBeenCalled();
await watcher.stopWatching();
expect(connection.removeOnLogsListener).toHaveBeenCalledTimes(1);
expect(connection.removeOnLogsListener).toHaveBeenCalledWith(12345);
});
});
@@ -327,6 +320,9 @@ describe('Real Solana Observation Mode Unit Tests', () => {
recordRpcRequestRate: vi.fn(),
recordQueueProcessed: vi.fn(),
recordRpcFetchError: vi.fn(),
recordWalletState: vi.fn(),
recordWalletIncomingRate: vi.fn(),
recordWalletProcessedRate: vi.fn(),
};
});
@@ -724,6 +720,75 @@ describe('Real Solana Observation Mode Unit Tests', () => {
loggerDebugSpy.mockRestore();
});
it('should correctly transition wallet state to HOT under overload and maintain NORMAL for others', async () => {
const { BoundedTransactionQueue } = await import('../src/solana/fetcher.js');
const queue = new BoundedTransactionQueue(
connection as any,
1, // maxConcurrency = 1
10, // minDelay = 10ms
5, // maxQueueSize = 5
mockMetrics,
);
queue.start();
// Push first item for walletA (goes active immediately)
queue.push({
signature: 'A1',
walletAddress: 'walletA',
slot: 100,
detectedAt: new Date(),
correlationId: 'corrA1',
onSuccess: async () => {},
onFailure: async () => {},
});
// walletA gets 4 more items queued (total depth 4/5 = 80%, should trigger HOT)
for (let i = 2; i <= 5; i++) {
queue.push({
signature: `A${i}`,
walletAddress: 'walletA',
slot: 100,
detectedAt: new Date(),
correlationId: `corrA${i}`,
onSuccess: async () => {},
onFailure: async () => {},
});
}
// walletB gets 1 item queued (depth 1/5, should stay NORMAL)
queue.push({
signature: 'B1',
walletAddress: 'walletB',
slot: 100,
detectedAt: new Date(),
correlationId: 'corrB1',
onSuccess: async () => {},
onFailure: async () => {},
});
expect(queue.getWalletState('walletA')).toBe('HOT');
expect(queue.getWalletState('walletB')).toBe('NORMAL');
queue.stop();
});
it('should pre-filter obviously irrelevant transactions using the lightweight log filter', async () => {
const { isLogIrrelevant } = await import('../src/solana/parser.js');
const irrelevantLogs = [
'Program Vote111111111111111111111111111111111111111 invoke [1]',
'Program Vote111111111111111111111111111111111111111 success',
];
const normalLogs = [
'Program ComputeBudget111111111111111111111111111111 invoke [1]',
'Program swapInstruction invoke [2]',
];
expect(isLogIrrelevant(irrelevantLogs)).toBe(true);
expect(isLogIrrelevant(normalLogs)).toBe(false);
});
it('should classify fetched transactions correctly based on SOL/SPL token balance changes', async () => {
const { isTransactionRelevant } = await import('../src/solana/parser.js');