chore: Implement ClearcutSender HTTP transport for telemetry

Implements the telemetry transport layer per the design doc:
- Ring buffer with overflow handling (1000 events max)
- HTTP transport using native fetch API with 30s timeout
- Daisy-chain flush scheduling (15min default interval)
- Transient vs permanent error classification (5xx/429 retry, 4xx drop)
- Server-side rate limiting via next_request_wait_millis
- Shutdown handling with 5s timeout for final flush

Author: ergunsh

src/telemetry/types.ts

@@ -47,6 +47,14 @@ export interface LogRequest {
   }>;
 }
 
+export interface LogResponse {
+  /**
+   * If present, the client must wait this many milliseconds before
+   * issuing the next HTTP request.
+   */
+  next_request_wait_millis?: number;
+}
+
 // Enums
 export enum OsType {
   OS_TYPE_UNSPECIFIED = 0,

src/telemetry/watchdog/clearcut-sender.ts

@@ -7,15 +7,42 @@
 import crypto from 'node:crypto';
 
 import {logger} from '../../logger.js';
-import type {ChromeDevToolsMcpExtension, OsType} from '../types.js';
+import type {
+  ChromeDevToolsMcpExtension,
+  LogRequest,
+  LogResponse,
+  OsType,
+} from '../types.js';
 
+const INCLUDE_PID_HEADER = process.env.INCLUDE_PID_HEADER === 'true';
+const MAX_BUFFER_SIZE = parseInt(process.env.MAX_BUFFER_SIZE ?? '', 10) || 1000;
+const FLUSH_INTERVAL_MS =
+  parseInt(process.env.FORCE_FLUSH_INTERVAL ?? '', 10) || 15 * 60 * 1000;
+const CLEARCUT_ENDPOINT =
+  process.env.CLEARCUT_ENDPOINT ??
+  'https://play.googleapis.com/log?format=json_proto';
+
+const LOG_SOURCE = 2839;
+const CLIENT_TYPE = 47;
+const MIN_RATE_LIMIT_WAIT_MS = 30_000;
+const REQUEST_TIMEOUT_MS = 30_000;
+const SHUTDOWN_TIMEOUT_MS = 5_000;
 const SESSION_ROTATION_INTERVAL_MS = 24 * 60 * 60 * 1000;
 
+interface BufferedEvent {
+  event: ChromeDevToolsMcpExtension;
+  timestamp: number;
+}
+
 export class ClearcutSender {
   #appVersion: string;
   #osType: OsType;
   #sessionId: string;
   #sessionCreated: number;
+  #buffer: BufferedEvent[] = [];
+  #flushTimer: ReturnType<typeof setTimeout> | null = null;
+  #isFlushing = false;
+  #timerStarted = false;
 
   constructor(appVersion: string, osType: OsType) {
     this.#appVersion = appVersion;
@@ -24,36 +51,187 @@ export class ClearcutSender {
     this.#sessionCreated = Date.now();
   }
 
-  async send(event: ChromeDevToolsMcpExtension): Promise<void> {
-    this.#rotateSessionIfNeeded();
-    const enrichedEvent = this.#enrichEvent(event);
-    this.transport(enrichedEvent);
-  }
+  enqueueEvent(event: ChromeDevToolsMcpExtension): void {
+    if (Date.now() - this.#sessionCreated > SESSION_ROTATION_INTERVAL_MS) {
+      this.#sessionId = crypto.randomUUID();
+      this.#sessionCreated = Date.now();
+    }
+
+    this.#addToBuffer({
+      ...event,
+      session_id: this.#sessionId,
+      app_version: this.#appVersion,
+      os_type: this.#osType,
+    });
 
-  transport(event: ChromeDevToolsMcpExtension): void {
-    logger('Telemetry event', JSON.stringify(event, null, 2));
+    if (!this.#timerStarted) {
+      this.#timerStarted = true;
+      this.#scheduleFlush(FLUSH_INTERVAL_MS);
+    }
   }
 
   async sendShutdownEvent(): Promise<void> {
+    if (this.#flushTimer) {
+      clearTimeout(this.#flushTimer);
+      this.#flushTimer = null;
+    }
+
     const shutdownEvent: ChromeDevToolsMcpExtension = {
       server_shutdown: {},
     };
-    await this.send(shutdownEvent);
+    this.enqueueEvent(shutdownEvent);
+
+    try {
+      await Promise.race([
+        this.#finalFlush(),
+        new Promise(resolve => setTimeout(resolve, SHUTDOWN_TIMEOUT_MS)),
+      ]);
+    } catch (error) {
+      logger('Final flush failed:', error);
+    }
   }
 
-  #rotateSessionIfNeeded(): void {
-    if (Date.now() - this.#sessionCreated > SESSION_ROTATION_INTERVAL_MS) {
-      this.#sessionId = crypto.randomUUID();
-      this.#sessionCreated = Date.now();
+  async #flush(): Promise<void> {
+    if (this.#isFlushing) {
+      return;
+    }
+
+    if (this.#buffer.length === 0) {
+      this.#scheduleFlush(FLUSH_INTERVAL_MS);
+      return;
+    }
+
+    this.#isFlushing = true;
+    let nextDelayMs = FLUSH_INTERVAL_MS;
+
+    // Optimistically remove events from buffer before sending.
+    // This prevents race conditions where a simultaneous #finalFlush would include these same events.
+    const eventsToSend = [...this.#buffer];
+    this.#buffer = [];
+
+    try {
+      const result = await this.#sendBatch(eventsToSend);
+
+      if (result.success) {
+        if (result.nextRequestWaitMs !== undefined) {
+          nextDelayMs = Math.max(
+            result.nextRequestWaitMs,
+            MIN_RATE_LIMIT_WAIT_MS,
+          );
+        }
+      } else if (result.isPermanentError) {
+        logger(
+          'Permanent error, dropped batch of',
+          eventsToSend.length,
+          'events',
+        );
+      } else {
+        // Transient error: Requeue events at the front of the buffer
+        // to maintain order and retry them later.
+        this.#buffer = [...eventsToSend, ...this.#buffer];
+      }
+    } catch (error) {
+      // Safety catch for unexpected errors, requeue events
+      this.#buffer = [...eventsToSend, ...this.#buffer];
+      logger('Flush failed unexpectedly:', error);
+    } finally {
+      this.#isFlushing = false;
+      this.#scheduleFlush(nextDelayMs);
     }
   }
 
-  #enrichEvent(event: ChromeDevToolsMcpExtension): ChromeDevToolsMcpExtension {
-    return {
-      ...event,
-      session_id: this.#sessionId,
-      app_version: this.#appVersion,
-      os_type: this.#osType,
+  #addToBuffer(event: ChromeDevToolsMcpExtension): void {
+    if (this.#buffer.length >= MAX_BUFFER_SIZE) {
+      this.#buffer.shift();
+      logger('Telemetry buffer overflow: dropped oldest event');
+    }
+    this.#buffer.push({
+      event,
+      timestamp: Date.now(),
+    });
+  }
+
+  #scheduleFlush(delayMs: number): void {
+    if (this.#flushTimer) {
+      clearTimeout(this.#flushTimer);
+    }
+    this.#flushTimer = setTimeout(() => {
+      this.#flush().catch(err => {
+        logger('Flush error:', err);
+      });
+    }, delayMs);
+  }
+
+  async #sendBatch(events: BufferedEvent[]): Promise<{
+    success: boolean;
+    isPermanentError?: boolean;
+    nextRequestWaitMs?: number;
+  }> {
+    const requestBody: LogRequest = {
+      log_source: LOG_SOURCE,
+      request_time_ms: Date.now().toString(),
+      client_info: {
+        client_type: CLIENT_TYPE,
+      },
+      log_event: events.map(({event, timestamp}) => ({
+        event_time_ms: timestamp.toString(),
+        source_extension_json: JSON.stringify(event),
+      })),
     };
+
+    const controller = new AbortController();
+    const timeoutId = setTimeout(() => controller.abort(), REQUEST_TIMEOUT_MS);
+    try {
+      const response = await fetch(CLEARCUT_ENDPOINT, {
+        method: 'POST',
+        headers: {
+          'Content-Type': 'application/json',
+          // Used in E2E tests to confirm that the watchdog process is killed
+          ...(INCLUDE_PID_HEADER ? {'X-Watchdog-Pid': process.pid.toString()} : {}),
+        },
+        body: JSON.stringify(requestBody),
+        signal: controller.signal,
+      });
+
+      clearTimeout(timeoutId);
+      if (response.ok) {
+        const data = (await response.json()) as LogResponse;
+        return {
+          success: true,
+          nextRequestWaitMs: data.next_request_wait_millis,
+        };
+      }
+
+      const status = response.status;
+      if (status >= 500 || status === 429) {
+        return {success: false};
+      }
+
+      logger('Telemetry permanent error:', status);
+      return {success: false, isPermanentError: true};
+    } catch {
+      clearTimeout(timeoutId);
+      return {success: false};
+    }
+  }
+
+  async #finalFlush(): Promise<void> {
+    if (this.#buffer.length === 0) {
+      return;
+    }
+    const eventsToSend = [...this.#buffer];
+    await this.#sendBatch(eventsToSend);
+  }
+
+  stopForTesting(): void {
+    if (this.#flushTimer) {
+      clearTimeout(this.#flushTimer);
+      this.#flushTimer = null;
+    }
+    this.#timerStarted = false;
+  }
+
+  get bufferSizeForTesting(): number {
+    return this.#buffer.length;
   }
 }

src/telemetry/watchdog/main.ts

@@ -107,9 +107,7 @@ function main() {
 
       const msg = JSON.parse(line);
       if (msg.type === WatchdogMessageType.LOG_EVENT && msg.payload) {
-        sender.send(msg.payload).catch(err => {
-          logger('Error sending event', err);
-        });
+        sender.enqueueEvent(msg.payload);
       }
     } catch (err) {
       logger('Failed to parse IPC message', err);