From 09c785f485138f3560da79aa4f61c414ba5c32bd Mon Sep 17 00:00:00 2001 From: Max Techera Date: Thu, 18 Dec 2025 14:11:25 -0300 Subject: [PATCH 1/8] perf(AGENT-581): add adaptive rate limiting to billing sync - Start with 100ms delay, increase exponentially on 429 errors - Reduce delay by 20% after 5 consecutive successful calls - Max delay capped at 5 seconds - Add heartbeat logging during long delays - Improve progress logging with current delay info --- .../billing/langfuse/LangfuseProvider.ts | 143 ++++++++++++++---- 1 file changed, 117 insertions(+), 26 deletions(-) diff --git a/packages/server/src/aai-utils/billing/langfuse/LangfuseProvider.ts b/packages/server/src/aai-utils/billing/langfuse/LangfuseProvider.ts index 3f4ef77ce91..a54a7a9c241 100644 --- a/packages/server/src/aai-utils/billing/langfuse/LangfuseProvider.ts +++ b/packages/server/src/aai-utils/billing/langfuse/LangfuseProvider.ts @@ -26,6 +26,75 @@ export class LangfuseProvider { // Future timestamp buffer (5 minutes) for timestamp validation private static readonly FUTURE_TIMESTAMP_BUFFER_SECONDS = 300 + // Adaptive rate limiter state + private adaptiveDelay = { + current: 100, // Start fast (100ms) + min: 100, // Minimum delay + max: 5000, // Maximum delay (5 seconds) + backoffMultiplier: 2, // Double on 429 + recoveryRate: 0.8, // Reduce by 20% after success + lastRateLimitTime: 0, // Track when we last hit a rate limit + consecutiveSuccesses: 0 // Track successful calls for faster recovery + } + + /** + * Record a successful API call - gradually reduce delay + */ + private recordSuccess(): void { + this.adaptiveDelay.consecutiveSuccesses++ + // After 5 consecutive successes, start reducing delay + if (this.adaptiveDelay.consecutiveSuccesses >= 5) { + this.adaptiveDelay.current = Math.max( + this.adaptiveDelay.min, + Math.floor(this.adaptiveDelay.current * this.adaptiveDelay.recoveryRate) + ) + this.adaptiveDelay.consecutiveSuccesses = 0 + } + } + + /** + * Record a rate limit hit - increase delay exponentially + */ + private recordRateLimit(): void { + this.adaptiveDelay.consecutiveSuccesses = 0 + this.adaptiveDelay.current = Math.min(this.adaptiveDelay.max, this.adaptiveDelay.current * this.adaptiveDelay.backoffMultiplier) + this.adaptiveDelay.lastRateLimitTime = Date.now() + log.info('Rate limit detected, increasing delay', { + newDelayMs: this.adaptiveDelay.current, + maxDelayMs: this.adaptiveDelay.max + }) + } + + /** + * Get current adaptive delay (in ms) + */ + private getAdaptiveDelay(): number { + return this.adaptiveDelay.current + } + + /** + * Wait with heartbeat logging for long delays + */ + private async waitWithHeartbeat(delayMs: number, context: string): Promise { + const HEARTBEAT_INTERVAL = 3000 // Log every 3 seconds + let elapsed = 0 + + while (elapsed < delayMs) { + const waitTime = Math.min(HEARTBEAT_INTERVAL, delayMs - elapsed) + await new Promise((resolve) => setTimeout(resolve, waitTime)) + elapsed += waitTime + + // Log heartbeat for long waits + if (elapsed < delayMs && delayMs > HEARTBEAT_INTERVAL) { + log.debug('Rate limit cooldown...', { + context, + elapsed: `${elapsed}ms`, + remaining: `${delayMs - elapsed}ms` + }) + } + } + } + /** * Metadata filter to exclude already-processed traces * Backward compatible: Traces without billing_status field are included (treated as != 'processed') @@ -46,10 +115,10 @@ export class LangfuseProvider { /** * Make authenticated request to Langfuse API with retry for rate limits + * Uses adaptive rate limiting - adjusts delays based on 429 responses */ private async fetchFromLangfuseAPI(endpoint: string, params: Record = {}, retryCount = 0): Promise { const MAX_RETRIES = BILLING_CONFIG.SYNC.MAX_RETRIES - const BASE_DELAY_MS = BILLING_CONFIG.SYNC.RETRY_DELAY_MS try { const url = `${this.langfuseBaseUrl}/api/public${endpoint}` @@ -60,20 +129,25 @@ export class LangfuseProvider { 'Content-Type': 'application/json' } }) + // Track successful call for adaptive rate limiting + this.recordSuccess() return response.data } catch (error: any) { const status = error.response?.status - // Retry on 429 (rate limit) with exponential backoff + // Retry on 429 (rate limit) with adaptive backoff if (status === 429 && retryCount < MAX_RETRIES) { - const delay = BASE_DELAY_MS * Math.pow(2, retryCount) // Exponential: 1s, 2s, 4s + // Record rate limit for adaptive throttling + this.recordRateLimit() + const delay = this.getAdaptiveDelay() * Math.pow(2, retryCount) log.warn('Rate limited by Langfuse, retrying...', { endpoint, retryCount: retryCount + 1, maxRetries: MAX_RETRIES, - delayMs: delay + delayMs: delay, + adaptiveDelay: this.adaptiveDelay.current }) - await new Promise((resolve) => setTimeout(resolve, delay)) + await this.waitWithHeartbeat(delay, `retry ${retryCount + 1}`) return this.fetchFromLangfuseAPI(endpoint, params, retryCount + 1) } @@ -378,9 +452,8 @@ export class LangfuseProvider { skippedCount += firstPageSkipped failedTraces.push(...firstPageResponse.failedEvents) - // Process remaining pages in batches + // Process remaining pages in batches with adaptive rate limiting const PAGE_BATCH_SIZE = BILLING_CONFIG.SYNC.PAGE_BATCH_SIZE - const RATE_LIMIT_DELAY_MS = BILLING_CONFIG.SYNC.RATE_LIMIT_DELAY_MS for (let startPage = 2; startPage <= totalPages; startPage += PAGE_BATCH_SIZE) { const endPage = Math.min(startPage + PAGE_BATCH_SIZE - 1, totalPages) @@ -388,7 +461,8 @@ export class LangfuseProvider { log.info('Processing page group', { startPage, endPage, - progress: `${endPage}/${totalPages}` + progress: `${endPage}/${totalPages}`, + adaptiveDelay: `${this.getAdaptiveDelay()}ms` }) const { billable: traces, skippedCount: pageSkipped } = await this.fetchPageGroup( @@ -404,9 +478,10 @@ export class LangfuseProvider { skippedCount += pageSkipped failedTraces.push(...response.failedEvents) - // Rate limit between batches + // Use adaptive delay between batches if (endPage < totalPages) { - await new Promise((resolve) => setTimeout(resolve, RATE_LIMIT_DELAY_MS)) + const delay = this.getAdaptiveDelay() + await new Promise((resolve) => setTimeout(resolve, delay)) } } @@ -447,6 +522,7 @@ export class LangfuseProvider { /** * Fetch and filter traces from multiple pages sequentially * Sequential fetching prevents ClickHouse database overload + * Uses adaptive rate limiting to optimize throughput */ private async fetchPageGroup( startPage: number, @@ -456,12 +532,21 @@ export class LangfuseProvider { traceId?: string ): Promise<{ billable: Trace[]; skippedCount: number }> { // Fetch pages sequentially to avoid ClickHouse overload - // IMPORTANT: Construct and await each request inside the loop - // to prevent all HTTP requests from firing in parallel + // Uses adaptive delays that adjust based on 429 responses const responses = [] - const PAGE_FETCH_DELAY_MS = BILLING_CONFIG.SYNC.PAGE_FETCH_DELAY_MS + const totalPages = endPage - startPage + 1 for (let page = startPage; page <= endPage; page++) { + const pageIndex = page - startPage + 1 + + // Log progress for longer fetches + if (totalPages > 2 && pageIndex % 2 === 0) { + log.debug('Fetching pages', { + progress: `${pageIndex}/${totalPages}`, + currentDelay: `${this.getAdaptiveDelay()}ms` + }) + } + // Execute fetch call and await immediately (truly sequential) const response = await this.fetchTraces({ fromTimestamp: fromTimestamp.toISOString(), @@ -474,9 +559,11 @@ export class LangfuseProvider { }) responses.push(response) - // Delay between pages to give ClickHouse breathing room + // Use adaptive delay between pages (lower minimum for page fetches) if (page < endPage) { - await new Promise((resolve) => setTimeout(resolve, PAGE_FETCH_DELAY_MS)) + // Use half the trace delay for page fetches (pages are lighter) + const delay = Math.max(50, Math.floor(this.getAdaptiveDelay() / 2)) + await new Promise((resolve) => setTimeout(resolve, delay)) } } @@ -505,11 +592,11 @@ export class LangfuseProvider { log.info('Starting trace processing', { totalTraces: filteredData.length, - referenceTime: nowUtc.toISOString() + referenceTime: nowUtc.toISOString(), + initialDelay: `${this.getAdaptiveDelay()}ms` }) - // Process traces sequentially to avoid rate limits - const TRACE_DELAY_MS = BILLING_CONFIG.SYNC.RATE_LIMIT_DELAY_MS + // Process traces sequentially with adaptive delays const startTime = Date.now() let successCount = 0 let failCount = 0 @@ -524,22 +611,25 @@ export class LangfuseProvider { successCount++ totalCredits += result.creditsData.credits.total || 0 - // Log every 10th trace or significant credits - if ((i + 1) % 10 === 0 || result.creditsData.credits.total > 100) { - log.info('Trace processed', { + // Log every 5th trace for better visibility (reduced from 10) + if ((i + 1) % 5 === 0 || result.creditsData.credits.total > 100) { + const elapsed = ((Date.now() - startTime) / 1000).toFixed(1) + log.info('Trace batch progress', { progress: `${i + 1}/${filteredData.length}`, - traceId: trace.id.substring(0, 8), credits: result.creditsData.credits.total, - runningTotal: totalCredits + runningTotal: totalCredits, + elapsed: `${elapsed}s`, + currentDelay: `${this.getAdaptiveDelay()}ms` }) } } else { failCount++ } - // Apply delay between traces (except last one) + // Apply adaptive delay between traces (except last one) if (i < filteredData.length - 1) { - await new Promise((resolve) => setTimeout(resolve, TRACE_DELAY_MS)) + const delay = this.getAdaptiveDelay() + await new Promise((resolve) => setTimeout(resolve, delay)) } } @@ -549,7 +639,8 @@ export class LangfuseProvider { failed: failCount, totalCredits, elapsedSeconds: elapsedSec, - avgSecondsPerTrace: (parseFloat(elapsedSec) / filteredData.length).toFixed(2) + avgSecondsPerTrace: filteredData.length > 0 ? (parseFloat(elapsedSec) / filteredData.length).toFixed(2) : '0', + finalDelay: `${this.getAdaptiveDelay()}ms` }) return processedData From 150890e4c7d98c1420e9a04d178d9268d61802a9 Mon Sep 17 00:00:00 2001 From: Max Techera Date: Thu, 18 Dec 2025 14:32:22 -0300 Subject: [PATCH 2/8] perf(AGENT-581): add 10-concurrent parallel trace processing - Add runWithConcurrency pool helper (10 concurrent workers) - Process traces in parallel instead of sequentially - Each worker respects adaptive delays between API calls - Progress logs show rate (traces/sec) and completion % - Expected ~10x throughput improvement when not rate limited --- .../billing/langfuse/LangfuseProvider.ts | 114 +++++++++++++----- 1 file changed, 87 insertions(+), 27 deletions(-) diff --git a/packages/server/src/aai-utils/billing/langfuse/LangfuseProvider.ts b/packages/server/src/aai-utils/billing/langfuse/LangfuseProvider.ts index a54a7a9c241..b1954996bc1 100644 --- a/packages/server/src/aai-utils/billing/langfuse/LangfuseProvider.ts +++ b/packages/server/src/aai-utils/billing/langfuse/LangfuseProvider.ts @@ -95,6 +95,61 @@ export class LangfuseProvider { } } + // Concurrency configuration + private static readonly MAX_CONCURRENCY = 10 + + /** + * Run tasks with limited concurrency (pool pattern) + * Processes items in parallel while respecting rate limits + */ + private async runWithConcurrency( + items: T[], + processor: (item: T, index: number) => Promise, + onProgress?: (completed: number, total: number, result: R | null) => void + ): Promise<(R | null)[]> { + const results: (R | null)[] = new Array(items.length).fill(null) + let nextIndex = 0 + let completed = 0 + + const processNext = async (): Promise => { + while (nextIndex < items.length) { + const currentIndex = nextIndex++ + const item = items[currentIndex] + + try { + // Apply adaptive delay before processing + const delay = this.getAdaptiveDelay() + if (delay > 0 && currentIndex > 0) { + await new Promise((resolve) => setTimeout(resolve, delay)) + } + + const result = await processor(item, currentIndex) + results[currentIndex] = result + completed++ + + if (onProgress) { + onProgress(completed, items.length, result) + } + } catch (error: any) { + completed++ + results[currentIndex] = null + log.error('Error in concurrent task', { + index: currentIndex, + error: error.message + }) + } + } + } + + // Start concurrent workers + const workers = Array(Math.min(LangfuseProvider.MAX_CONCURRENCY, items.length)) + .fill(null) + .map(() => processNext()) + + await Promise.all(workers) + return results + } + /** * Metadata filter to exclude already-processed traces * Backward compatible: Traces without billing_status field are included (treated as != 'processed') @@ -584,62 +639,67 @@ export class LangfuseProvider { private async convertUsageToCredits(usageData: Trace[]): Promise> { const validTraces = await Promise.all(usageData.map((trace) => this.validateUsageData(trace))) const filteredData = usageData.filter((_, index) => validTraces[index]) - const processedData: Array<{ creditsData: CreditsData; fullTrace: any }> = [] // Use UTC timestamp for consistency const nowUtc = new Date() const nowUtcSeconds = Math.floor(nowUtc.getTime() / 1000) - log.info('Starting trace processing', { + log.info('Starting parallel trace processing', { totalTraces: filteredData.length, + concurrency: LangfuseProvider.MAX_CONCURRENCY, referenceTime: nowUtc.toISOString(), initialDelay: `${this.getAdaptiveDelay()}ms` }) - // Process traces sequentially with adaptive delays + // Process traces in parallel with controlled concurrency const startTime = Date.now() let successCount = 0 let failCount = 0 let totalCredits = 0 + let lastLogTime = Date.now() - for (let i = 0; i < filteredData.length; i++) { - const trace = filteredData[i] - const result = await this.processTrace(trace, nowUtcSeconds) - - if (result) { - processedData.push(result) - successCount++ - totalCredits += result.creditsData.credits.total || 0 - - // Log every 5th trace for better visibility (reduced from 10) - if ((i + 1) % 5 === 0 || result.creditsData.credits.total > 100) { - const elapsed = ((Date.now() - startTime) / 1000).toFixed(1) - log.info('Trace batch progress', { - progress: `${i + 1}/${filteredData.length}`, - credits: result.creditsData.credits.total, + const results = await this.runWithConcurrency( + filteredData, + async (trace) => { + return this.processTrace(trace, nowUtcSeconds) + }, + (completed, total, result) => { + if (result) { + successCount++ + totalCredits += result.creditsData.credits.total || 0 + } else { + failCount++ + } + + // Log progress every 2 seconds or every 10 completions + const now = Date.now() + if (now - lastLogTime > 2000 || completed % 10 === 0) { + lastLogTime = now + const elapsed = ((now - startTime) / 1000).toFixed(1) + log.info('Parallel processing progress', { + completed: `${completed}/${total}`, + success: successCount, + failed: failCount, runningTotal: totalCredits, elapsed: `${elapsed}s`, + rate: `${(completed / parseFloat(elapsed)).toFixed(1)}/s`, currentDelay: `${this.getAdaptiveDelay()}ms` }) } - } else { - failCount++ } + ) - // Apply adaptive delay between traces (except last one) - if (i < filteredData.length - 1) { - const delay = this.getAdaptiveDelay() - await new Promise((resolve) => setTimeout(resolve, delay)) - } - } + // Collect successful results + const processedData = results.filter((r): r is { creditsData: CreditsData; fullTrace: any } => r !== null) const elapsedSec = ((Date.now() - startTime) / 1000).toFixed(1) - log.info('Trace processing complete', { + log.info('Parallel trace processing complete', { success: successCount, failed: failCount, totalCredits, elapsedSeconds: elapsedSec, avgSecondsPerTrace: filteredData.length > 0 ? (parseFloat(elapsedSec) / filteredData.length).toFixed(2) : '0', + effectiveRate: `${(filteredData.length / parseFloat(elapsedSec)).toFixed(1)}/s`, finalDelay: `${this.getAdaptiveDelay()}ms` }) From f4736f925c3c3ed7e43f7fe3cfafb74deaa60e2c Mon Sep 17 00:00:00 2001 From: Max Techera Date: Thu, 18 Dec 2025 17:18:05 -0300 Subject: [PATCH 3/8] perf(AGENT-581): start with zero delay until 429 hit MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Start with 0ms delay for maximum throughput - On first 429, jump to 200ms base delay - Subsequent 429s double the delay (200 → 400 → 800 → ...) - Recovery still reduces delay back toward 0ms --- .../aai-utils/billing/langfuse/LangfuseProvider.ts | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/packages/server/src/aai-utils/billing/langfuse/LangfuseProvider.ts b/packages/server/src/aai-utils/billing/langfuse/LangfuseProvider.ts index b1954996bc1..9d896e3d81d 100644 --- a/packages/server/src/aai-utils/billing/langfuse/LangfuseProvider.ts +++ b/packages/server/src/aai-utils/billing/langfuse/LangfuseProvider.ts @@ -26,10 +26,10 @@ export class LangfuseProvider { // Future timestamp buffer (5 minutes) for timestamp validation private static readonly FUTURE_TIMESTAMP_BUFFER_SECONDS = 300 - // Adaptive rate limiter state + // Adaptive rate limiter state - starts with NO delay, only throttles on 429 private adaptiveDelay = { - current: 100, // Start fast (100ms) - min: 100, // Minimum delay + current: 0, // Start with NO delay - maximum throughput + min: 0, // Allow zero delay when no rate limits hit max: 5000, // Maximum delay (5 seconds) backoffMultiplier: 2, // Double on 429 recoveryRate: 0.8, // Reduce by 20% after success @@ -57,7 +57,11 @@ export class LangfuseProvider { */ private recordRateLimit(): void { this.adaptiveDelay.consecutiveSuccesses = 0 - this.adaptiveDelay.current = Math.min(this.adaptiveDelay.max, this.adaptiveDelay.current * this.adaptiveDelay.backoffMultiplier) + // If current is 0, start with 200ms base delay; otherwise double it + const BASE_DELAY_ON_429 = 200 + const newDelay = + this.adaptiveDelay.current === 0 ? BASE_DELAY_ON_429 : this.adaptiveDelay.current * this.adaptiveDelay.backoffMultiplier + this.adaptiveDelay.current = Math.min(this.adaptiveDelay.max, newDelay) this.adaptiveDelay.lastRateLimitTime = Date.now() log.info('Rate limit detected, increasing delay', { newDelayMs: this.adaptiveDelay.current, From 8f88a663461c7dc34919308e564fa88562a9ae99 Mon Sep 17 00:00:00 2001 From: Max Techera Date: Thu, 18 Dec 2025 17:22:18 -0300 Subject: [PATCH 4/8] perf(AGENT-581): remove forced delays between page fetches - Remove Math.max(50, ...) forcing minimum 50ms between pages - Skip delay entirely when adaptive delay is 0 - Pages now fetch as fast as API allows when not rate limited --- .../billing/langfuse/LangfuseProvider.ts | 15 +++++++++------ 1 file changed, 9 insertions(+), 6 deletions(-) diff --git a/packages/server/src/aai-utils/billing/langfuse/LangfuseProvider.ts b/packages/server/src/aai-utils/billing/langfuse/LangfuseProvider.ts index 9d896e3d81d..d4813e3cf1b 100644 --- a/packages/server/src/aai-utils/billing/langfuse/LangfuseProvider.ts +++ b/packages/server/src/aai-utils/billing/langfuse/LangfuseProvider.ts @@ -537,10 +537,12 @@ export class LangfuseProvider { skippedCount += pageSkipped failedTraces.push(...response.failedEvents) - // Use adaptive delay between batches + // Use adaptive delay between batches (no delay when not rate limited) if (endPage < totalPages) { const delay = this.getAdaptiveDelay() - await new Promise((resolve) => setTimeout(resolve, delay)) + if (delay > 0) { + await new Promise((resolve) => setTimeout(resolve, delay)) + } } } @@ -618,11 +620,12 @@ export class LangfuseProvider { }) responses.push(response) - // Use adaptive delay between pages (lower minimum for page fetches) + // Use adaptive delay between pages (no delay when not rate limited) if (page < endPage) { - // Use half the trace delay for page fetches (pages are lighter) - const delay = Math.max(50, Math.floor(this.getAdaptiveDelay() / 2)) - await new Promise((resolve) => setTimeout(resolve, delay)) + const delay = this.getAdaptiveDelay() + if (delay > 0) { + await new Promise((resolve) => setTimeout(resolve, delay)) + } } } From 44ee0c5eac8446c13f1acd61e9cdff1da8b1213c Mon Sep 17 00:00:00 2001 From: Max Techera Date: Thu, 18 Dec 2025 18:34:55 -0300 Subject: [PATCH 5/8] fix(AGENT-581): skip already-processed traces after full fetch - Check billing_status on full trace data (fresh from API) - Skip trace before expensive operations (costs, Stripe) - Catches stale Langfuse cache causing duplicate processing --- .../aai-utils/billing/langfuse/LangfuseProvider.ts | 13 ++++++++++++- 1 file changed, 12 insertions(+), 1 deletion(-) diff --git a/packages/server/src/aai-utils/billing/langfuse/LangfuseProvider.ts b/packages/server/src/aai-utils/billing/langfuse/LangfuseProvider.ts index d4813e3cf1b..5350ec6be4e 100644 --- a/packages/server/src/aai-utils/billing/langfuse/LangfuseProvider.ts +++ b/packages/server/src/aai-utils/billing/langfuse/LangfuseProvider.ts @@ -728,11 +728,22 @@ export class LangfuseProvider { return undefined } + const fullTrace = await this.fetchTrace(trace.id) + + // Check if already processed (fresh data from full trace fetch) + // This catches traces where Langfuse cache was stale during initial fetch + const fullMetadata = fullTrace?.metadata as any + if (fullMetadata?.billing_status === 'processed') { + log.debug('Skipping already processed trace (detected on full fetch)', { + traceId: trace.id + }) + return undefined + } + const metadata = { ...((trace.metadata || {}) as TraceMetadata), aiCredentialsOwnership: 'user' } as TraceMetadata - const fullTrace = await this.fetchTrace(trace.id) // TODO: Update calculateCosts, getModelUsage, and buildCreditsData to work with v4 API response types const costs = await this.calculateCosts(fullTrace as any) metadata.aiCredentialsOwnership = costs.aiCredentialsOwnership From 48179048b3613cff06d2b536dcb39b3055b09ad3 Mon Sep 17 00:00:00 2001 From: Max Techera Date: Fri, 19 Dec 2025 11:47:51 -0300 Subject: [PATCH 6/8] fix(AGENT-581): filter out undefined results from parallel processing - processTrace returns undefined for skipped traces - Filter was checking !== null, missing undefined values - Changed to != null to catch both null and undefined --- .../server/src/aai-utils/billing/langfuse/LangfuseProvider.ts | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/packages/server/src/aai-utils/billing/langfuse/LangfuseProvider.ts b/packages/server/src/aai-utils/billing/langfuse/LangfuseProvider.ts index 5350ec6be4e..93cf3421204 100644 --- a/packages/server/src/aai-utils/billing/langfuse/LangfuseProvider.ts +++ b/packages/server/src/aai-utils/billing/langfuse/LangfuseProvider.ts @@ -696,8 +696,8 @@ export class LangfuseProvider { } ) - // Collect successful results - const processedData = results.filter((r): r is { creditsData: CreditsData; fullTrace: any } => r !== null) + // Collect successful results (filter out null AND undefined) + const processedData = results.filter((r): r is { creditsData: CreditsData; fullTrace: any } => r != null) const elapsedSec = ((Date.now() - startTime) / 1000).toFixed(1) log.info('Parallel trace processing complete', { From 649028e15654619df50356f0cb6bd7cb9d28b9a9 Mon Sep 17 00:00:00 2001 From: Max Techera Date: Fri, 19 Dec 2025 13:20:31 -0300 Subject: [PATCH 7/8] fix(AGENT-581): reduce memory usage in billing sync - Replace fullTrace objects (~500KB each) with traceContext (~1KB) containing only timestamp and metadata needed for updates - Replace meterEvents and processedTraces arrays with counters to prevent unbounded accumulation during large syncs - Reduces memory from ~100MB to ~5MB for 10K trace sync --- .../billing/langfuse/LangfuseProvider.ts | 25 ++++++++-- .../billing/stripe/StripeProvider.ts | 47 ++++++++++--------- 2 files changed, 46 insertions(+), 26 deletions(-) diff --git a/packages/server/src/aai-utils/billing/langfuse/LangfuseProvider.ts b/packages/server/src/aai-utils/billing/langfuse/LangfuseProvider.ts index 93cf3421204..2269e7cf50e 100644 --- a/packages/server/src/aai-utils/billing/langfuse/LangfuseProvider.ts +++ b/packages/server/src/aai-utils/billing/langfuse/LangfuseProvider.ts @@ -388,7 +388,7 @@ export class LangfuseProvider { return await stripeProvider.syncUsageToStripe( creditsDataWithTraces.map((item) => ({ ...item.creditsData, - fullTrace: item.fullTrace + traceContext: item.traceContext })) ) } @@ -643,7 +643,9 @@ export class LangfuseProvider { private async validateUsageData(trace: Trace): Promise { return !!(trace.id && typeof trace.totalCost === 'number' && typeof trace.latency === 'number') } - private async convertUsageToCredits(usageData: Trace[]): Promise> { + private async convertUsageToCredits( + usageData: Trace[] + ): Promise> { const validTraces = await Promise.all(usageData.map((trace) => this.validateUsageData(trace))) const filteredData = usageData.filter((_, index) => validTraces[index]) @@ -697,7 +699,9 @@ export class LangfuseProvider { ) // Collect successful results (filter out null AND undefined) - const processedData = results.filter((r): r is { creditsData: CreditsData; fullTrace: any } => r != null) + const processedData = results.filter( + (r): r is { creditsData: CreditsData; traceContext: { timestamp: string; metadata: any } } => r != null + ) const elapsedSec = ((Date.now() - startTime) / 1000).toFixed(1) log.info('Parallel trace processing complete', { @@ -713,7 +717,10 @@ export class LangfuseProvider { return processedData } - private async processTrace(trace: Trace, nowUtcSeconds: number): Promise<{ creditsData: CreditsData; fullTrace: any } | undefined> { + private async processTrace( + trace: Trace, + nowUtcSeconds: number + ): Promise<{ creditsData: CreditsData; traceContext: { timestamp: string; metadata: any } } | undefined> { try { const traceDate = new Date(trace.timestamp) const traceTimestampSeconds = Math.floor(traceDate.getTime() / 1000) @@ -751,7 +758,15 @@ export class LangfuseProvider { const modelUsage = await this.getModelUsage(fullTrace as any) const creditsData = this.buildCreditsData(fullTrace as any, metadata, costs, credits, modelUsage, traceTimestampSeconds) - return { creditsData, fullTrace: fullTrace } + // Only store minimal context needed for metadata updates (timestamp + metadata) + // This reduces memory from ~500KB to ~1KB per trace + return { + creditsData, + traceContext: { + timestamp: fullTrace.timestamp, + metadata: fullTrace.metadata + } + } } catch (error: any) { log.error('Error processing trace', { traceId: trace.id, error: error.message }) return undefined diff --git a/packages/server/src/aai-utils/billing/stripe/StripeProvider.ts b/packages/server/src/aai-utils/billing/stripe/StripeProvider.ts index 3d9879db6dd..73abb91ff8f 100644 --- a/packages/server/src/aai-utils/billing/stripe/StripeProvider.ts +++ b/packages/server/src/aai-utils/billing/stripe/StripeProvider.ts @@ -335,10 +335,12 @@ export class StripeProvider { } } - async syncUsageToStripe(creditsData: Array): Promise<{ + async syncUsageToStripe(creditsData: Array): Promise<{ meterEvents: Stripe.Billing.MeterEvent[] failedEvents: Array<{ traceId: string; error: string }> processedTraces: string[] + meterEventCount: number + processedCount: number }> { try { log.info('Syncing usage to Stripe', { count: creditsData.length }) @@ -349,9 +351,10 @@ export class StripeProvider { const BATCH_SIZE = BILLING_CONFIG.VALIDATION.MAX_BATCH_SIZE const DELAY_BETWEEN_BATCHES = BILLING_CONFIG.VALIDATION.BATCH_DELAY_MS - const meterEvents: Stripe.Billing.MeterEvent[] = [] + // Use counters instead of arrays to prevent memory accumulation + let meterEventCount = 0 + let processedCount = 0 const failedEvents: Array<{ traceId: string; error: string }> = [] - const processedTraces: string[] = [] let selfHealedCount = 0 let duplicateEventCount = 0 let adjustedTimestampCount = 0 @@ -498,11 +501,11 @@ export class StripeProvider { selfHealedCount++ } - // Only add to meterEvents if not a duplicate + // Only count if not a duplicate if (!isDuplicate) { - meterEvents.push(result) + meterEventCount++ } - processedTraces.push(data.traceId) + processedCount++ break } catch (error) { retryCount++ @@ -544,8 +547,8 @@ export class StripeProvider { if (selfHealedCount > 0) { log.info('Self-healing: Processed untagged traces', { count: selfHealedCount, - totalProcessed: processedTraces.length, - percentage: ((selfHealedCount / processedTraces.length) * 100).toFixed(2) + '%' + totalProcessed: processedCount, + percentage: ((selfHealedCount / processedCount) * 100).toFixed(2) + '%' }) } @@ -553,8 +556,8 @@ export class StripeProvider { if (duplicateEventCount > 0) { log.info('Duplicate events handled gracefully', { count: duplicateEventCount, - totalProcessed: processedTraces.length, - percentage: ((duplicateEventCount / processedTraces.length) * 100).toFixed(2) + '%', + totalProcessed: processedCount, + percentage: ((duplicateEventCount / processedCount) * 100).toFixed(2) + '%', note: 'These traces were already in Stripe but not marked as processed in Langfuse' }) } @@ -563,22 +566,24 @@ export class StripeProvider { if (adjustedTimestampCount > 0) { log.info('Historical data: Adjusted timestamps for Stripe 35-day limitation', { count: adjustedTimestampCount, - totalProcessed: processedTraces.length, - percentage: ((adjustedTimestampCount / processedTraces.length) * 100).toFixed(2) + '%', + totalProcessed: processedCount, + percentage: ((adjustedTimestampCount / processedCount) * 100).toFixed(2) + '%', note: 'Traces older than 35 days were batched to 34 days ago with original dates preserved in metadata' }) } // Final flush to ensure all metadata updates are persisted log.info('Final flush of Langfuse metadata updates', { - totalProcessed: processedTraces.length + totalProcessed: processedCount }) await this.langfuseV3.flushAsync() return { - meterEvents, + meterEvents: [], // Empty array for API compatibility - use meterEventCount instead failedEvents, - processedTraces + processedTraces: [], // Empty array for API compatibility - use processedCount instead + meterEventCount, + processedCount } } catch (error) { log.error('Error syncing usage to Stripe', { error }) @@ -592,7 +597,7 @@ export class StripeProvider { } } - private validateUsageEvent(data: CreditsData & { fullTrace: any }): string | null { + private validateUsageEvent(data: CreditsData & { traceContext: { timestamp: string; metadata: any } }): string | null { try { // Check required metadata fields for (const field of BILLING_CONFIG.METADATA_FIELDS.REQUIRED) { @@ -632,7 +637,7 @@ export class StripeProvider { } private async updateTraceMetadata( - data: CreditsData & { fullTrace: any }, + data: CreditsData & { traceContext: { timestamp: string; metadata: any } }, result: Stripe.Billing.MeterEvent, batchStartTime: number, batchSize: number, @@ -646,9 +651,9 @@ export class StripeProvider { // Using v3 client for this operation as v4 doesn't have trace metadata update method yet await this.langfuseV3.trace({ id: data.traceId, - timestamp: data.fullTrace?.timestamp, + timestamp: data.traceContext?.timestamp ? new Date(data.traceContext.timestamp) : undefined, metadata: { - ...data.fullTrace?.metadata, + ...data.traceContext?.metadata, billing_status: 'processed', meter_event_id: result.identifier, billing_details: { @@ -690,7 +695,7 @@ export class StripeProvider { log.debug('Trace metadata updated successfully', { traceId: data.traceId, - billingStatus: 'processed', + billing_status: data.traceContext?.metadata?.billing_status, meterEventId: result.identifier }) @@ -710,7 +715,7 @@ export class StripeProvider { private processBatchResults( batchResults: PromiseSettledResult[], - batch: Array, + batch: Array, meterEvents: Stripe.Billing.MeterEvent[], failedEvents: Array<{ traceId: string; error: string }>, processedTraces: string[] From f13cd5996e733adb73528bf30f79b708cb7b0183 Mon Sep 17 00:00:00 2001 From: Max Techera Date: Tue, 10 Feb 2026 10:25:55 -0300 Subject: [PATCH 8/8] fix(AGENT-581): fix counter reads, dead code, and division-by-zero in billing sync - Read processedCount/failedCount from StripeProvider instead of empty arrays - Add explicit return type to processAndSyncTraces - Remove dead lastRateLimitTime field and processBatchResults method - Guard all rate/percentage calculations against division by zero - Fix stale billing_status in debug log --- .../billing/langfuse/LangfuseProvider.ts | 45 ++++++++++++------- .../billing/stripe/StripeProvider.ts | 28 +----------- 2 files changed, 31 insertions(+), 42 deletions(-) diff --git a/packages/server/src/aai-utils/billing/langfuse/LangfuseProvider.ts b/packages/server/src/aai-utils/billing/langfuse/LangfuseProvider.ts index 2269e7cf50e..b0eaf77d4fc 100644 --- a/packages/server/src/aai-utils/billing/langfuse/LangfuseProvider.ts +++ b/packages/server/src/aai-utils/billing/langfuse/LangfuseProvider.ts @@ -33,7 +33,6 @@ export class LangfuseProvider { max: 5000, // Maximum delay (5 seconds) backoffMultiplier: 2, // Double on 429 recoveryRate: 0.8, // Reduce by 20% after success - lastRateLimitTime: 0, // Track when we last hit a rate limit consecutiveSuccesses: 0 // Track successful calls for faster recovery } @@ -62,7 +61,6 @@ export class LangfuseProvider { const newDelay = this.adaptiveDelay.current === 0 ? BASE_DELAY_ON_429 : this.adaptiveDelay.current * this.adaptiveDelay.backoffMultiplier this.adaptiveDelay.current = Math.min(this.adaptiveDelay.max, newDelay) - this.adaptiveDelay.lastRateLimitTime = Date.now() log.info('Rate limit detected, increasing delay', { newDelayMs: this.adaptiveDelay.current, maxDelayMs: this.adaptiveDelay.max @@ -377,20 +375,30 @@ export class LangfuseProvider { /** * Convert traces to credits and sync to Stripe */ - private async processAndSyncTraces(traces: Trace[]) { + private async processAndSyncTraces(traces: Trace[]): Promise<{ + processedCount: number + failedCount: number + failedEvents: Array<{ traceId: string; error: string }> + }> { if (traces.length === 0) { - return { processedTraces: [], failedEvents: [], meterEvents: [] } + return { processedCount: 0, failedCount: 0, failedEvents: [] } } const creditsDataWithTraces = await this.convertUsageToCredits(traces) const stripeProvider = new StripeProvider() - return await stripeProvider.syncUsageToStripe( + const result = await stripeProvider.syncUsageToStripe( creditsDataWithTraces.map((item) => ({ ...item.creditsData, traceContext: item.traceContext })) ) + + return { + processedCount: result.processedCount, + failedCount: result.failedEvents.length, + failedEvents: result.failedEvents + } } /** @@ -435,9 +443,12 @@ export class LangfuseProvider { const response = await this.processAndSyncTraces(traces) return { - processedTraces: response.processedTraces, + processedTraces: [], failedTraces: response.failedEvents, - skippedTraces: [] + skippedTraces: [], + processedCount: response.processedCount, + failedCount: response.failedCount, + skippedCount: 0 } } @@ -506,8 +517,8 @@ export class LangfuseProvider { const { billable: firstPageTraces, skippedCount: firstPageSkipped } = this.filterBillableTraces(initialResponse.data) const firstPageResponse = await this.processAndSyncTraces(firstPageTraces) - processedCount += firstPageResponse.processedTraces.length - failedCount += firstPageResponse.failedEvents.length + processedCount += firstPageResponse.processedCount + failedCount += firstPageResponse.failedCount skippedCount += firstPageSkipped failedTraces.push(...firstPageResponse.failedEvents) @@ -532,8 +543,8 @@ export class LangfuseProvider { ) const response = await this.processAndSyncTraces(traces) - processedCount += response.processedTraces.length - failedCount += response.failedEvents.length + processedCount += response.processedCount + failedCount += response.failedCount skippedCount += pageSkipped failedTraces.push(...response.failedEvents) @@ -684,14 +695,15 @@ export class LangfuseProvider { const now = Date.now() if (now - lastLogTime > 2000 || completed % 10 === 0) { lastLogTime = now - const elapsed = ((now - startTime) / 1000).toFixed(1) + const elapsedMs = now - startTime + const elapsed = (elapsedMs / 1000).toFixed(1) log.info('Parallel processing progress', { completed: `${completed}/${total}`, success: successCount, failed: failCount, runningTotal: totalCredits, elapsed: `${elapsed}s`, - rate: `${(completed / parseFloat(elapsed)).toFixed(1)}/s`, + rate: elapsedMs > 0 ? `${(completed / (elapsedMs / 1000)).toFixed(1)}/s` : 'N/A', currentDelay: `${this.getAdaptiveDelay()}ms` }) } @@ -703,14 +715,15 @@ export class LangfuseProvider { (r): r is { creditsData: CreditsData; traceContext: { timestamp: string; metadata: any } } => r != null ) - const elapsedSec = ((Date.now() - startTime) / 1000).toFixed(1) + const elapsedMs = Date.now() - startTime + const elapsedSec = (elapsedMs / 1000).toFixed(1) log.info('Parallel trace processing complete', { success: successCount, failed: failCount, totalCredits, elapsedSeconds: elapsedSec, - avgSecondsPerTrace: filteredData.length > 0 ? (parseFloat(elapsedSec) / filteredData.length).toFixed(2) : '0', - effectiveRate: `${(filteredData.length / parseFloat(elapsedSec)).toFixed(1)}/s`, + avgSecondsPerTrace: filteredData.length > 0 ? (elapsedMs / 1000 / filteredData.length).toFixed(2) : '0', + effectiveRate: elapsedMs > 0 ? `${(filteredData.length / (elapsedMs / 1000)).toFixed(1)}/s` : 'N/A', finalDelay: `${this.getAdaptiveDelay()}ms` }) diff --git a/packages/server/src/aai-utils/billing/stripe/StripeProvider.ts b/packages/server/src/aai-utils/billing/stripe/StripeProvider.ts index 73abb91ff8f..497b6a0c6fb 100644 --- a/packages/server/src/aai-utils/billing/stripe/StripeProvider.ts +++ b/packages/server/src/aai-utils/billing/stripe/StripeProvider.ts @@ -567,7 +567,7 @@ export class StripeProvider { log.info('Historical data: Adjusted timestamps for Stripe 35-day limitation', { count: adjustedTimestampCount, totalProcessed: processedCount, - percentage: ((adjustedTimestampCount / processedCount) * 100).toFixed(2) + '%', + percentage: processedCount > 0 ? ((adjustedTimestampCount / processedCount) * 100).toFixed(2) + '%' : 'N/A', note: 'Traces older than 35 days were batched to 34 days ago with original dates preserved in metadata' }) } @@ -695,7 +695,7 @@ export class StripeProvider { log.debug('Trace metadata updated successfully', { traceId: data.traceId, - billing_status: data.traceContext?.metadata?.billing_status, + billing_status: 'processed', meterEventId: result.identifier }) @@ -713,30 +713,6 @@ export class StripeProvider { } } - private processBatchResults( - batchResults: PromiseSettledResult[], - batch: Array, - meterEvents: Stripe.Billing.MeterEvent[], - failedEvents: Array<{ traceId: string; error: string }>, - processedTraces: string[] - ): void { - batchResults.forEach((result, index) => { - if (result.status === 'fulfilled') { - meterEvents.push(result.value.result) - processedTraces.push(result.value.traceId) - } else { - const error = result.reason - // Only add to failedEvents if it's not a resource_missing error that was handled - if (!(error.code === 'resource_missing' && error.param === 'payload[stripe_customer_id]')) { - failedEvents.push({ - traceId: batch[index].traceId, - error: error?.message || 'Unknown error during meter event creation' - }) - } - } - }) - } - async getMeterEventSummaries( customerId: string, startTime?: number,