From ef3aeb3c589c2ba3e632af195c43a594c223d5b9 Mon Sep 17 00:00:00 2001 From: daniel-lxs Date: Fri, 15 Aug 2025 12:33:29 -0500 Subject: [PATCH] fix: address PR review feedback for telemetry queue persistence - Make console.log statements conditional based on debug flag - Add singleton reset method for TelemetryQueueManager - Fix race condition in persistQueue with debouncing and promise tracking - Improve error handling in PostHogTelemetryClient to differentiate error types - Make retry interval configurable in QueuedTelemetryClient - Update tests to reflect 24-hour event expiration instead of retry-based removal - Add test for handling corrupted JSON queue files - Fix test expectations for retry count and event filtering --- .../telemetry/src/PostHogTelemetryClient.ts | 30 +++- .../telemetry/src/QueuedTelemetryClient.ts | 21 ++- .../telemetry/src/TelemetryQueueManager.ts | 150 ++++++++++++++---- .../__tests__/TelemetryQueueManager.test.ts | 38 +++-- 4 files changed, 188 insertions(+), 51 deletions(-) diff --git a/packages/telemetry/src/PostHogTelemetryClient.ts b/packages/telemetry/src/PostHogTelemetryClient.ts index bf6e1fb80f..60e5283f52 100644 --- a/packages/telemetry/src/PostHogTelemetryClient.ts +++ b/packages/telemetry/src/PostHogTelemetryClient.ts @@ -93,7 +93,35 @@ export class PostHogTelemetryClient extends QueuedTelemetryClient { if (this.debug) { console.error(`[PostHogTelemetryClient#sendEvent] Failed to send event: ${event.event}`, error) } - // Re-throw to trigger our queuing mechanism + + // Differentiate between different types of errors + const errorMessage = error instanceof Error ? error.message : String(error) + + // Check if it's a network error or other transient issue + const _isNetworkError = + errorMessage.toLowerCase().includes("network") || + errorMessage.toLowerCase().includes("timeout") || + errorMessage.toLowerCase().includes("econnrefused") || + errorMessage.toLowerCase().includes("enotfound") || + errorMessage.toLowerCase().includes("fetch") + + // Check if it's a configuration error that won't be fixed by retrying + const isConfigError = + errorMessage.toLowerCase().includes("api key") || + errorMessage.toLowerCase().includes("invalid configuration") + + if (isConfigError) { + // Don't queue config errors - they won't succeed on retry + if (this.debug) { + console.error( + `[PostHogTelemetryClient#sendEvent] Configuration error, not queuing: ${errorMessage}`, + ) + } + // Silently fail for config errors to not break the extension + return + } + + // Re-throw network and other transient errors to trigger queuing throw error } } diff --git a/packages/telemetry/src/QueuedTelemetryClient.ts b/packages/telemetry/src/QueuedTelemetryClient.ts index 3fd16c25fb..cdd2c5fd35 100644 --- a/packages/telemetry/src/QueuedTelemetryClient.ts +++ b/packages/telemetry/src/QueuedTelemetryClient.ts @@ -11,18 +11,27 @@ export abstract class QueuedTelemetryClient extends BaseTelemetryClient { protected clientId: string private retryTimer: NodeJS.Timeout | null = null private isOnline = true - private readonly RETRY_CHECK_INTERVAL = 30000 // Check for retries every 30 seconds + private readonly retryCheckInterval: number - constructor(clientId: string, storagePath: string, subscription?: TelemetryEventSubscription, debug = false) { + constructor( + clientId: string, + storagePath: string, + subscription?: TelemetryEventSubscription, + debug = false, + retryCheckInterval = 30000, // Default: Check for retries every 30 seconds + ) { super(subscription, debug) this.clientId = clientId + this.retryCheckInterval = retryCheckInterval // Initialize queue manager try { - this.queueManager = TelemetryQueueManager.getInstance(storagePath) + this.queueManager = TelemetryQueueManager.getInstance(storagePath, debug) this.startRetryTimer() } catch (error) { - console.error(`Failed to initialize queue manager: ${error}`) + if (debug) { + console.error(`Failed to initialize queue manager: ${error}`) + } } } @@ -110,14 +119,14 @@ export abstract class QueuedTelemetryClient extends BaseTelemetryClient { */ private startRetryTimer(): void { if (this.debug) { - console.info(`[${this.clientId}] Starting retry timer, checking every ${this.RETRY_CHECK_INTERVAL}ms`) + console.info(`[${this.clientId}] Starting retry timer, checking every ${this.retryCheckInterval}ms`) } this.retryTimer = setInterval(() => { if (this.debug) { console.info(`[${this.clientId}] Retry timer triggered, checking for events to retry`) } this.processQueuedEvents() - }, this.RETRY_CHECK_INTERVAL) + }, this.retryCheckInterval) } /** diff --git a/packages/telemetry/src/TelemetryQueueManager.ts b/packages/telemetry/src/TelemetryQueueManager.ts index a0ab41e2ca..cebcabf76c 100644 --- a/packages/telemetry/src/TelemetryQueueManager.ts +++ b/packages/telemetry/src/TelemetryQueueManager.ts @@ -30,10 +30,16 @@ export class TelemetryQueueManager { private cleanupInterval: NodeJS.Timeout | null = null private readonly QUEUE_VERSION = 1 private readonly MAX_EVENT_AGE = 24 * 60 * 60 * 1000 // 24 hours instead of 7 days + private debug = false + private persistPromise: Promise | null = null + private pendingPersist = false - private constructor(storagePath: string) { + private constructor(storagePath: string, debug = false) { + this.debug = debug || process.env.DEBUG_TELEMETRY === "true" this.persistPath = path.join(storagePath, "telemetry-queue.json") - console.log(`[TelemetryQueue] Initializing with path: ${this.persistPath}`) + if (this.debug) { + console.log(`[TelemetryQueue] Initializing with path: ${this.persistPath}`) + } this.loadQueue() this.startPeriodicFlush() this.startPeriodicCleanup() @@ -42,26 +48,40 @@ export class TelemetryQueueManager { /** * Get or create the singleton instance */ - public static getInstance(storagePath: string): TelemetryQueueManager { + public static getInstance(storagePath: string, debug = false): TelemetryQueueManager { if (!this.instance) { - this.instance = new TelemetryQueueManager(storagePath) + this.instance = new TelemetryQueueManager(storagePath, debug) } return this.instance } + /** + * Reset the singleton instance (for testing) + */ + public static resetInstance(): void { + if (this.instance) { + this.instance.shutdown() + this.instance = null + } + } + /** * Add an event to the queue */ public enqueue(event: TelemetryEvent, clientId: string): void { - console.log(`[TelemetryQueue] Enqueueing event: ${event.event} for client: ${clientId}`) + if (this.debug) { + console.log(`[TelemetryQueue] Enqueueing event: ${event.event} for client: ${clientId}`) + } // Don't queue if we've reached the maximum size if (this.queue.length >= this.maxQueueSize) { // Remove oldest events to make room (FIFO) const removed = this.queue.shift() - console.log( - `[TelemetryQueue] Queue full (${this.maxQueueSize}), removed oldest event: ${removed?.event.event}`, - ) + if (this.debug) { + console.log( + `[TelemetryQueue] Queue full (${this.maxQueueSize}), removed oldest event: ${removed?.event.event}`, + ) + } } const queuedEvent: QueuedEvent = { @@ -72,8 +92,10 @@ export class TelemetryQueueManager { } this.queue.push(queuedEvent) - console.log(`[TelemetryQueue] Queue size after enqueue: ${this.queue.length}`) - this.persistQueue() + if (this.debug) { + console.log(`[TelemetryQueue] Queue size after enqueue: ${this.queue.length}`) + } + this.schedulePersist() } /** @@ -89,23 +111,31 @@ export class TelemetryQueueManager { try { const clientEvents = this.queue.filter((e) => e.clientId === clientId) - console.log(`[TelemetryQueue] Processing ${clientEvents.length} events for client: ${clientId}`) + if (this.debug) { + console.log(`[TelemetryQueue] Processing ${clientEvents.length} events for client: ${clientId}`) + } for (const queuedEvent of clientEvents) { try { - console.log( - `[TelemetryQueue] Attempting to send event: ${queuedEvent.event.event}, retry count: ${queuedEvent.retryCount}`, - ) + if (this.debug) { + console.log( + `[TelemetryQueue] Attempting to send event: ${queuedEvent.event.event}, retry count: ${queuedEvent.retryCount}`, + ) + } await sendFunction(queuedEvent.event) // Remove successfully sent event - console.log(`[TelemetryQueue] Successfully sent event: ${queuedEvent.event.event}`) + if (this.debug) { + console.log(`[TelemetryQueue] Successfully sent event: ${queuedEvent.event.event}`) + } this.removeEvent(queuedEvent) } catch (error) { // Increment retry count queuedEvent.retryCount++ - console.log( - `[TelemetryQueue] Failed to send event: ${queuedEvent.event.event}, retry count now: ${queuedEvent.retryCount}, error: ${error}`, - ) + if (this.debug) { + console.log( + `[TelemetryQueue] Failed to send event: ${queuedEvent.event.event}, retry count now: ${queuedEvent.retryCount}, error: ${error}`, + ) + } // Don't remove based on retry count - let it keep trying until 24 hours } @@ -168,33 +198,75 @@ export class TelemetryQueueManager { const originalCount = state.events.length this.queue = state.events.filter((e) => e.timestamp > cutoffTime) - console.log( - `[TelemetryQueue] Loaded ${this.queue.length} events from disk (filtered ${originalCount - this.queue.length} old events)`, - ) + if (this.debug) { + console.log( + `[TelemetryQueue] Loaded ${this.queue.length} events from disk (filtered ${originalCount - this.queue.length} old events)`, + ) + } // If we filtered out any events, persist the cleaned queue if (this.queue.length < originalCount) { - this.persistQueue() + this.schedulePersist() } } } catch (error) { // File doesn't exist or is corrupted, start with empty queue - console.log(`[TelemetryQueue] No existing queue file found or error loading: ${error}`) + if (this.debug) { + console.log(`[TelemetryQueue] No existing queue file found or error loading: ${error}`) + } this.queue = [] } } + /** + * Schedule a persist operation with debouncing to avoid race conditions + */ + private schedulePersist(): void { + if (this.pendingPersist) { + // A persist is already scheduled + return + } + + this.pendingPersist = true + + // Use setImmediate to batch multiple rapid enqueue operations + setImmediate(() => { + this.pendingPersist = false + this.persistQueue() + }) + } + /** * Persist queue to disk */ private async persistQueue(): Promise { + // If a persist is already in progress, wait for it to complete + if (this.persistPromise) { + await this.persistPromise + return + } + + this.persistPromise = this.doPersist() + try { + await this.persistPromise + } finally { + this.persistPromise = null + } + } + + /** + * Actually perform the persist operation + */ + private async doPersist(): Promise { try { const state: QueueState = { events: this.queue, version: this.QUEUE_VERSION, } - console.log(`[TelemetryQueue] Persisting ${this.queue.length} events to disk`) + if (this.debug) { + console.log(`[TelemetryQueue] Persisting ${this.queue.length} events to disk`) + } // Ensure directory exists const dir = path.dirname(this.persistPath) @@ -205,10 +277,14 @@ export class TelemetryQueueManager { await fs.writeFile(tempPath, JSON.stringify(state, null, 2)) await fs.rename(tempPath, this.persistPath) - console.log(`[TelemetryQueue] Successfully persisted queue to: ${this.persistPath}`) + if (this.debug) { + console.log(`[TelemetryQueue] Successfully persisted queue to: ${this.persistPath}`) + } } catch (error) { // Log error but don't throw - telemetry should not break the app - console.error("[TelemetryQueue] Failed to persist telemetry queue:", error) + if (this.debug) { + console.error("[TelemetryQueue] Failed to persist telemetry queue:", error) + } } } @@ -243,14 +319,16 @@ export class TelemetryQueueManager { */ private performAggressiveCleanup(): void { const originalSize = this.queue.length - console.log(`[TelemetryQueue] Running aggressive cleanup, current queue size: ${originalSize}`) + if (this.debug) { + console.log(`[TelemetryQueue] Running aggressive cleanup, current queue size: ${originalSize}`) + } // Remove old events const cutoffTime = Date.now() - this.MAX_EVENT_AGE const beforeOldFilter = this.queue.length this.queue = this.queue.filter((e) => e.timestamp > cutoffTime) const removedOld = beforeOldFilter - this.queue.length - if (removedOld > 0) { + if (removedOld > 0 && this.debug) { console.log(`[TelemetryQueue] Removed ${removedOld} events older than 24 hours`) } @@ -262,15 +340,19 @@ export class TelemetryQueueManager { this.queue.sort((a, b) => b.timestamp - a.timestamp) const beforeTrim = this.queue.length this.queue = this.queue.slice(0, this.maxQueueSize) - console.log(`[TelemetryQueue] Trimmed queue from ${beforeTrim} to ${this.maxQueueSize} events`) + if (this.debug) { + console.log(`[TelemetryQueue] Trimmed queue from ${beforeTrim} to ${this.maxQueueSize} events`) + } } // Persist if we made changes if (this.queue.length !== originalSize) { - console.log( - `[TelemetryQueue] Cleanup complete, queue size changed from ${originalSize} to ${this.queue.length}`, - ) - this.persistQueue() + if (this.debug) { + console.log( + `[TelemetryQueue] Cleanup complete, queue size changed from ${originalSize} to ${this.queue.length}`, + ) + } + this.schedulePersist() } } @@ -283,7 +365,7 @@ export class TelemetryQueueManager { this.queue = this.queue.filter((e) => e.timestamp > cutoffTime) if (this.queue.length < originalLength) { - this.persistQueue() + this.schedulePersist() } } diff --git a/packages/telemetry/src/__tests__/TelemetryQueueManager.test.ts b/packages/telemetry/src/__tests__/TelemetryQueueManager.test.ts index b1a898add5..3dce238165 100644 --- a/packages/telemetry/src/__tests__/TelemetryQueueManager.test.ts +++ b/packages/telemetry/src/__tests__/TelemetryQueueManager.test.ts @@ -130,11 +130,13 @@ describe("TelemetryQueueManager", () => { const sendFunction = vi.fn().mockRejectedValue(new Error("Network error")) await queueManager.processQueue("test-client", sendFunction) - const events = queueManager.getEventsForRetry("test-client") - expect(events[0].retryCount).toBe(1) + // Access the internal queue directly to check retry count + // eslint-disable-next-line @typescript-eslint/no-explicit-any + const internalQueue = (queueManager as any).queue + expect(internalQueue[0].retryCount).toBe(1) }) - it("should remove events after max retries", async () => { + it("should not remove events based on retry count (events expire after 24 hours)", async () => { const event: TelemetryEvent = { event: TelemetryEventName.TASK_CREATED, properties: { taskId: "test-123" }, @@ -142,16 +144,16 @@ describe("TelemetryQueueManager", () => { queueManager.enqueue(event, "test-client") - // Set retry count to max + // Set retry count to high value // eslint-disable-next-line @typescript-eslint/no-explicit-any const queuedEvents = (queueManager as any).queue - queuedEvents[0].retryCount = 4 + queuedEvents[0].retryCount = 10 const sendFunction = vi.fn().mockRejectedValue(new Error("Network error")) await queueManager.processQueue("test-client", sendFunction) const stats = queueManager.getStats() - expect(stats.queueSize).toBe(0) // Event removed after max retries + expect(stats.queueSize).toBe(1) // Event NOT removed, will expire after 24 hours }) }) @@ -163,6 +165,22 @@ describe("TelemetryQueueManager", () => { expect(queueManager.getRetryDelay(3)).toBe(8000) expect(queueManager.getRetryDelay(10)).toBe(60000) // Max delay }) + + describe("loadQueue with corrupted file", () => { + it("should handle corrupted JSON gracefully", async () => { + vi.mocked(fs.readFile).mockResolvedValueOnce("{ invalid json }") + + // Create new instance to trigger load + resetQueueManagerInstance() + queueManager = TelemetryQueueManager.getInstance(testStoragePath) + + // Wait for async load + await new Promise((resolve) => setTimeout(resolve, 10)) + + const stats = queueManager.getStats() + expect(stats.queueSize).toBe(0) // Should start with empty queue + }) + }) }) describe("shouldRetry", () => { @@ -250,19 +268,19 @@ describe("TelemetryQueueManager", () => { expect(stats.queueSize).toBe(1) }) - it("should filter out events older than 7 days", async () => { + it("should filter out events older than 24 hours", async () => { const savedQueue = { version: 1, events: [ { event: { event: TelemetryEventName.TASK_CREATED }, - timestamp: Date.now() - 8 * 24 * 60 * 60 * 1000, // 8 days old + timestamp: Date.now() - 25 * 60 * 60 * 1000, // 25 hours old retryCount: 0, clientId: "test-client", }, { event: { event: TelemetryEventName.TASK_COMPLETED }, - timestamp: Date.now() - 6 * 24 * 60 * 60 * 1000, // 6 days old + timestamp: Date.now() - 23 * 60 * 60 * 1000, // 23 hours old retryCount: 0, clientId: "test-client", }, @@ -279,7 +297,7 @@ describe("TelemetryQueueManager", () => { await new Promise((resolve) => setTimeout(resolve, 10)) const stats = queueManager.getStats() - expect(stats.queueSize).toBe(1) // Only the 6-day-old event + expect(stats.queueSize).toBe(1) // Only the 23-hour-old event }) })