add logs, fix a race condition with event sending

Author: nirinchev

src/common/logger.ts

@@ -35,6 +35,7 @@ export const LogId = {
     telemetryMetadataError: mongoLogId(1_002_005),
     deviceIdResolutionError: mongoLogId(1_002_006),
     deviceIdTimeout: mongoLogId(1_002_007),
+    telemetryClose: mongoLogId(1_002_008),
 
     toolExecute: mongoLogId(1_003_001),
     toolExecuteFailure: mongoLogId(1_003_002),

src/telemetry/eventCache.ts

@@ -34,11 +34,18 @@ export class EventCache {
     }
 
     /**
-     * Gets a copy of the currently cached events
+     * Gets the number of currently cached events
+     */
+    public get size(): number {
+        return this.cache.size;
+    }
+
+    /**
+     * Gets a copy of the currently cached events along with their ids
      * @returns Array of cached BaseEvent objects
      */
-    public getEvents(): BaseEvent[] {
-        return Array.from(this.cache.values());
+    public getEvents(): { id: number; event: BaseEvent }[] {
+        return Array.from(this.cache.entries()).map(([id, event]) => ({ id, event }));
     }
 
     /**
@@ -53,10 +60,11 @@ export class EventCache {
     }
 
     /**
-     * Clears all cached events
+     * Removes cached events by their ids
      */
-    public clearEvents(): void {
-        this.cache.clear();
-        this.nextId = 0;
+    public removeEvents(ids: number[]): void {
+        for (const id of ids) {
+            this.cache.delete(id);
+        }
     }
 }

src/telemetry/telemetry.ts

@@ -87,8 +87,27 @@ export class Telemetry {
     public async close(): Promise<void> {
         this.isBufferingEvents = false;
 
+        this.session.logger.debug({
+            id: LogId.telemetryClose,
+            message: `Closing telemetry and flushing ${this.eventCache.size} events`,
+            context: "telemetry",
+        });
+
         // Wait up to 5 seconds for events to be sent before closing, but don't throw if it times out
-        await Promise.race([new Promise((resolve) => setTimeout(resolve, 5000)), this.emit([])]);
+        const flushTimeout = 5000;
+        await Promise.race([
+            new Promise<void>((resolve) =>
+                setTimeout(() => {
+                    this.session.logger.debug({
+                        id: LogId.telemetryClose,
+                        message: `Failed to flush remaining events within ${flushTimeout}ms timeout`,
+                        context: "telemetry",
+                    });
+                    resolve();
+                }, flushTimeout)
+            ),
+            this.emit([]),
+        ]);
     }
 
     /**
@@ -151,7 +170,7 @@ export class Telemetry {
 
         try {
             const cachedEvents = this.eventCache.getEvents();
-            const allEvents = [...cachedEvents, ...events];
+            const allEvents = [...cachedEvents.map((e) => e.event), ...events];
 
             this.session.logger.debug({
                 id: LogId.telemetryEmitStart,
@@ -161,11 +180,11 @@ export class Telemetry {
 
             const result = await this.sendEvents(this.session.apiClient, allEvents);
             if (result.success) {
-                this.eventCache.clearEvents();
+                this.eventCache.removeEvents(cachedEvents.map((e) => e.id));
                 this.session.logger.debug({
                     id: LogId.telemetryEmitSuccess,
                     context: "telemetry",
-                    message: `Sent ${allEvents.length} events successfully: ${JSON.stringify(allEvents, null, 2)}`,
+                    message: `Sent ${allEvents.length} events successfully: ${JSON.stringify(allEvents)}`,
                 });
                 this.events.emit("events-emitted");
                 return;