chore: Implement ClearcutSender HTTP transport for telemetry disabled by default (#805)

This PR completes the telemetry system by implementing the transport
layer for `ClearcutSender`. It enables actual HTTP communication with
the Clearcut backend, handling event batching, rate limiting, and
reliable delivery, including robust shutdown handling.

**Key Changes:**
* **HTTP Transport**: Implemented `fetch`-based transport sending `POST`
requests to the Clearcut HTTP server.
* **Event Batching**: Events are now buffered and flushed periodically
(default: 15 minutes) or on shutdown.
*   **Reliability & Rate Limiting**:
* **Server-Side Backoff**: Respects `next_request_wait_millis` from
server responses to handle rate limiting dynamically.
* **Transient Error Retries**: Failed requests (5xx, 429) result in
events being requeued for the next flush.
* **Request Timeouts**: Enforced 30s timeout on requests to prevent
hanging processes.
* **Session Rotation**: Automatically rotates session IDs every 24
hours.
*   **Safety & Stability**:
* **Buffer Overflow Protection**: Caps the buffer at 1000 events to
prevent memory leaks, dropping oldest events if necessary.
* **Optimistic Removal**: Prevents race conditions and duplicate events
during shutdown by optimistically removing events from the buffer before
sending.
*   **Testing Improvements**:
* **E2E Robustness**: Updated E2E tests to use a mock web server instead
of relying on the logger to log specific lines.

**Implementation Roadmap:**
These changes finalize the planned telemetry architecture:
1. **CLI & Opt-out Mechanism
([Merged](#757
2. **Logger Scaffolding & Integration
([Merged](#758
3. **Persistence Layer
([Merged](#766
4. **Watchdog Process Architecture
([Merged](#769
5.  **Transport, Batching & Retries (This PR):**
* Finalized `ClearcutSender` with HTTP transport, batching, and
server-directed backoff strategies.

---------

Co-authored-by: Alex Rudenko <alexrudenko@chromium.org>

Author: ergunsh

src/cli.ts

@@ -209,7 +209,23 @@ export const cliOptions = {
     // Marked as `false` until the feature is ready to be enabled by default.
     default: false,
     hidden: true,
-    describe: 'Set to false to opt-out of usage statistics collection.',
+    describe:
+      'Set to false to opt-out of usage statistics collection. Google collects usage data to improve the tool, handled under the Google Privacy Policy (https://policies.google.com/privacy). This is independent from Chrome browser metrics.',
+  },
+  clearcutEndpoint: {
+    type: 'string',
+    hidden: true,
+    describe: 'Endpoint for Clearcut telemetry.',
+  },
+  clearcutForceFlushIntervalMs: {
+    type: 'number',
+    hidden: true,
+    describe: 'Force flush interval in milliseconds (for testing).',
+  },
+  clearcutIncludePidHeader: {
+    type: 'boolean',
+    hidden: true,
+    describe: 'Include watchdog PID in Clearcut request headers (for testing).',
   },
 } satisfies Record<string, YargsOptions>;
 

src/main.ts

@@ -42,6 +42,9 @@ if (args.usageStatistics) {
   clearcutLogger = new ClearcutLogger({
     logFile: args.logFile,
     appVersion: VERSION,
+    clearcutEndpoint: args.clearcutEndpoint,
+    clearcutForceFlushIntervalMs: args.clearcutForceFlushIntervalMs,
+    clearcutIncludePidHeader: args.clearcutIncludePidHeader,
   });
 }
 

src/telemetry/clearcut-logger.ts

@@ -37,6 +37,9 @@ export class ClearcutLogger {
     logFile?: string;
     persistence?: Persistence;
     watchdogClient?: WatchdogClient;
+    clearcutEndpoint?: string;
+    clearcutForceFlushIntervalMs?: number;
+    clearcutIncludePidHeader?: boolean;
   }) {
     this.#persistence = options.persistence ?? new FilePersistence();
     this.#watchdog =
@@ -46,6 +49,9 @@ export class ClearcutLogger {
         appVersion: options.appVersion,
         osType: detectOsType(),
         logFile: options.logFile,
+        clearcutEndpoint: options.clearcutEndpoint,
+        clearcutForceFlushIntervalMs: options.clearcutForceFlushIntervalMs,
+        clearcutIncludePidHeader: options.clearcutIncludePidHeader,
       });
   }
 

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-client.ts

@@ -20,6 +20,9 @@ export class WatchdogClient {
       appVersion: string;
       osType: OsType;
       logFile?: string;
+      clearcutEndpoint?: string;
+      clearcutForceFlushIntervalMs?: number;
+      clearcutIncludePidHeader?: boolean;
     },
     options?: {spawn?: typeof spawn},
   ) {
@@ -37,6 +40,17 @@ export class WatchdogClient {
     if (config.logFile) {
       args.push(`--log-file=${config.logFile}`);
     }
+    if (config.clearcutEndpoint) {
+      args.push(`--clearcut-endpoint=${config.clearcutEndpoint}`);
+    }
+    if (config.clearcutForceFlushIntervalMs) {
+      args.push(
+        `--clearcut-force-flush-interval-ms=${config.clearcutForceFlushIntervalMs}`,
+      );
+    }
+    if (config.clearcutIncludePidHeader) {
+      args.push('--clearcut-include-pid-header');
+    }
 
     const spawner = options?.spawn ?? spawn;
     this.#childProcess = spawner(process.execPath, args, {

src/telemetry/watchdog/clearcut-sender.ts

@@ -7,53 +7,246 @@
 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';
 
+export interface ClearcutSenderConfig {
+  appVersion: string;
+  osType: OsType;
+  clearcutEndpoint?: string;
+  forceFlushIntervalMs?: number;
+  includePidHeader?: boolean;
+}
+
+const MAX_BUFFER_SIZE = 1000;
+const DEFAULT_CLEARCUT_ENDPOINT =
+  'https://play.googleapis.com/log?format=json_proto';
+const DEFAULT_FLUSH_INTERVAL_MS = 15 * 60 * 1000;
+
+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;
+  #clearcutEndpoint: string;
+  #flushIntervalMs: number;
+  #includePidHeader: boolean;
   #sessionId: string;
   #sessionCreated: number;
+  #buffer: BufferedEvent[] = [];
+  #flushTimer: ReturnType<typeof setTimeout> | null = null;
+  #isFlushing = false;
+  #timerStarted = false;
 
-  constructor(appVersion: string, osType: OsType) {
-    this.#appVersion = appVersion;
-    this.#osType = osType;
+  constructor(config: ClearcutSenderConfig) {
+    this.#appVersion = config.appVersion;
+    this.#osType = config.osType;
+    this.#clearcutEndpoint =
+      config.clearcutEndpoint ?? DEFAULT_CLEARCUT_ENDPOINT;
+    this.#flushIntervalMs =
+      config.forceFlushIntervalMs ?? DEFAULT_FLUSH_INTERVAL_MS;
+    this.#includePidHeader = config.includePidHeader ?? false;
     this.#sessionId = crypto.randomUUID();
     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();
+    }
 
-  transport(event: ChromeDevToolsMcpExtension): void {
-    logger('Telemetry event', JSON.stringify(event, null, 2));
+    this.#addToBuffer({
+      ...event,
+      session_id: this.#sessionId,
+      app_version: this.#appVersion,
+      os_type: this.#osType,
+    });
+
+    if (!this.#timerStarted) {
+      this.#timerStarted = true;
+      this.#scheduleFlush(this.#flushIntervalMs);
+    }
   }
 
   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(this.#flushIntervalMs);
+      return;
+    }
+
+    this.#isFlushing = true;
+    let nextDelayMs = this.#flushIntervalMs;
+
+    // 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(this.#clearcutEndpoint, {
+        method: 'POST',
+        headers: {
+          'Content-Type': 'application/json',
+          // Used in E2E tests to confirm that the watchdog process is killed
+          ...(this.#includePidHeader
+            ? {'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;
   }
 }