diff --git a/apps/cli-docs/src/fragments/configuration.md b/apps/cli-docs/src/fragments/configuration.md index ca1235af1..19bdc04ed 100644 --- a/apps/cli-docs/src/fragments/configuration.md +++ b/apps/cli-docs/src/fragments/configuration.md @@ -92,6 +92,16 @@ export SENTRY_DISABLE_TIPS=1 # per-session, overrides the stored preference Only the tips are removed. Pagination hints, dry-run and cancellation notices, and the cache-age footer (`cached · 3m ago · use -f to refresh`) are still shown. +## Telemetry + +The CLI sends usage data and error reports about itself to Sentry. To opt out, set `SENTRY_CLI_NO_TELEMETRY=1` or `DO_NOT_TRACK=1`, or run `sentry cli defaults telemetry off`. + +When telemetry can't be sent, for example while offline or when a firewall blocks Sentry, the CLI keeps it in its local database and sends up to 10 stored envelopes on each later run. The cache holds at most 100 envelopes and 4 MiB, compressed. When it's full, the oldest usage data goes first, then the oldest error reports. It discards error reports after 30 days and everything else after 4 days. Metrics, such as timings, aren't cached: when they can't be sent, they're dropped. + +After the first failed send, the CLI stops sending for 5 minutes, so an unreachable Sentry delays a command's exit by about 3 seconds at most. Long-running commands, such as `sentry mcp` and `sentry local`, then try one send every 5 minutes, and send up to 10 stored envelopes once one succeeds. + +Opting out deletes the cache on the next run. The library (`createSentrySDK`) never caches telemetry. + ## Global Options These flags are accepted by every command. They are not shown in individual command `--help` output, but are always available. diff --git a/packages/cli/src/lib/db/schema.ts b/packages/cli/src/lib/db/schema.ts index 8176067e3..4f06e3fa4 100644 --- a/packages/cli/src/lib/db/schema.ts +++ b/packages/cli/src/lib/db/schema.ts @@ -18,12 +18,12 @@ import { logger } from "../logger.js"; import { storedCredentialIdentity } from "./identity-hash.js"; import type { Database } from "./sqlite.js"; -export const CURRENT_SCHEMA_VERSION = 20; +export const CURRENT_SCHEMA_VERSION = 21; /** Environment variable to disable auto-repair */ const NO_AUTO_REPAIR_ENV = "SENTRY_CLI_NO_AUTO_REPAIR"; -type SqliteType = "TEXT" | "INTEGER"; +type SqliteType = "TEXT" | "INTEGER" | "BLOB"; export type ColumnDef = { type: SqliteType; @@ -328,6 +328,23 @@ export const TABLE_SCHEMAS: Record = { result_count: { type: "INTEGER", notNull: true }, }, }, + /** + * Telemetry envelopes that failed to send, kept for a later run to retry + * (`src/lib/telemetry/offline-store.ts`). `envelope` holds the serialized + * envelope bytes; `size` is their length, for the byte cap. `is_error` is 1 + * when the envelope has an error event, which is kept longer and dropped + * last over the caps. Rows past their max age are deleted on read. Dropping + * the table only loses telemetry that may never have been delivered anyway. + */ + telemetry_envelope_queue: { + columns: { + id: { type: "INTEGER", primaryKey: true }, + created_at: { type: "INTEGER", notNull: true }, + is_error: { type: "INTEGER", notNull: true, default: "0" }, + size: { type: "INTEGER", notNull: true }, + envelope: { type: "BLOB", notNull: true }, + }, + }, }; /** Generate CREATE TABLE DDL from column definitions */ @@ -1172,6 +1189,12 @@ export function runMigrations(db: Database): void { db.exec(EXPECTED_TABLES.completion_cache as string); } + // Migration 20 -> 21: Add telemetry_envelope_queue for telemetry envelopes + // that failed to send, retried by the next run. + if (currentVersion < 21) { + db.exec(EXPECTED_TABLES.telemetry_envelope_queue as string); + } + if (currentVersion < CURRENT_SCHEMA_VERSION) { db.query("UPDATE schema_version SET version = ?").run( CURRENT_SCHEMA_VERSION, diff --git a/packages/cli/src/lib/mcp/server.ts b/packages/cli/src/lib/mcp/server.ts index 4c9becb70..7b1cf52d3 100644 --- a/packages/cli/src/lib/mcp/server.ts +++ b/packages/cli/src/lib/mcp/server.ts @@ -33,6 +33,12 @@ import { CLI_VERSION, SENTRY_CLI_DSN } from "../constants.js"; import { logger } from "../logger.js"; import { isTelemetryEnabled } from "../telemetry.js"; import { redactTelemetryEnvelope } from "../telemetry/credential-redaction.js"; +import { clearStoredEnvelopes } from "../telemetry/offline-store.js"; +import { + makeStoringTransport, + STORED_ENVELOPES_PER_RUN, + type StoringTransport, +} from "../telemetry/offline-transport.js"; import { makeNodeTransportWithDeadline } from "../telemetry/zstd-transport.js"; import type { McpServerConfig } from "./config.js"; import { reportCliError } from "../error-reporting.js"; @@ -245,6 +251,13 @@ export function initMcpTelemetry(config: McpServerConfig): void { "app.upstream.host": config.sentryHost, "app.url.full": config.mcpUrl, }; + // Unsent telemetry shares the CLI's store, so opting out clears it, as in + // `withTelemetry()`. + const storeTelemetry = isTelemetryEnabled(); + if (!storeTelemetry) { + clearStoredEnvelopes(); + } + let storingTransport: StoringTransport | undefined; // Always the CLI's own project, never a DSN from the environment: an MCP // server started from a project shell or IDE inherits the app's SENTRY_DSN. @@ -269,9 +282,17 @@ export function initMcpTelemetry(config: McpServerConfig): void { release: CLI_VERSION, // Same credential redaction and request deadline as the CLI's own // telemetry: a stalled ingest socket must not keep the process alive - // after the client disconnects (#1956). + // after the client disconnects (#1956). Envelopes that fail to send are + // stored like the CLI's own. transport: (transportOptions) => { - const transport = makeNodeTransportWithDeadline(transportOptions); + if (storeTelemetry) { + storingTransport = makeStoringTransport( + transportOptions, + makeNodeTransportWithDeadline, + ); + } + const transport = + storingTransport ?? makeNodeTransportWithDeadline(transportOptions); return { send: (envelope) => transport.send(redactTelemetryEnvelope(envelope)), flush: (timeout) => transport.flush(timeout), @@ -292,6 +313,9 @@ export function initMcpTelemetry(config: McpServerConfig): void { if (client) { setCurrentClient(client); } + // Retry what earlier runs failed to send; `close()` on disconnect waits + // for it. The breaker sends another batch whenever ingest recovers. + storingTransport?.sendStored(STORED_ENVELOPES_PER_RUN); } /** diff --git a/packages/cli/src/lib/telemetry.ts b/packages/cli/src/lib/telemetry.ts index 51af58026..c9abc9d57 100644 --- a/packages/cli/src/lib/telemetry.ts +++ b/packages/cli/src/lib/telemetry.ts @@ -48,6 +48,12 @@ import { } from "@sentry/toolkit-core/games"; import { getSentryBaseUrl, isSentrySaasUrl } from "./sentry-urls.js"; import { redactTelemetryEnvelope } from "./telemetry/credential-redaction.js"; +import { clearStoredEnvelopes } from "./telemetry/offline-store.js"; +import { + makeStoringTransport, + STORED_ENVELOPES_PER_RUN, + type StoringTransport, +} from "./telemetry/offline-transport.js"; import { makeCompressedTransport } from "./telemetry/zstd-transport.js"; export type { Span } from "@sentry/core"; @@ -215,9 +221,18 @@ export async function withTelemetry( const enabled = isTelemetryEnabled(); const client = initSentry(enabled, options); if (!client?.getOptions().enabled) { + // Opting out also discards telemetry stored by earlier runs. Library mode + // never touches the store: it belongs to the CLI, not the host. + if (!(enabled || options?.libraryMode)) { + clearStoredEnvelopes(); + } return await callback(undefined); } + // Retry telemetry that earlier runs failed to send. The sends run in the + // background; the exit flush in createBeforeExitHandler waits for them. + storingTransport?.sendStored(STORED_ENVELOPES_PER_RUN); + // Initialize user and instance context await initTelemetryContext(); @@ -494,6 +509,9 @@ const hasGetSystemErrorMap = (() => { /** Current beforeExit handler, tracked so it can be replaced on re-init */ let currentBeforeExitHandler: (() => void) | null = null; +/** Transport of the current client, unless library mode left the store out. */ +let storingTransport: StoringTransport | undefined; + let manualHttpTracing = false; /** @@ -665,8 +683,15 @@ export function initSentry( // smaller payloads, faster compress/decompress on both sides. // Automatic gzip fallback when running on Node < 22.15, where // `node:zlib`'s zstd support is unavailable. + // Outside library mode, envelopes that fail to send are stored for the + // next run (telemetry/offline-transport.ts). Library mode must not write + // the host's disk or add work to its process. transport: (transportOptions) => { - const transport = makeCompressedTransport(transportOptions); + storingTransport = libraryMode + ? undefined + : makeStoringTransport(transportOptions); + const transport = + storingTransport ?? makeCompressedTransport(transportOptions); return { // The SDK adds log scope attributes after beforeSendLog and skips // beforeSend for internal errors. Redact at the final delivery boundary. diff --git a/packages/cli/src/lib/telemetry/offline-store.ts b/packages/cli/src/lib/telemetry/offline-store.ts new file mode 100644 index 000000000..3bcf66d06 --- /dev/null +++ b/packages/cli/src/lib/telemetry/offline-store.ts @@ -0,0 +1,225 @@ +/** + * SQLite-backed `OfflineStore` for telemetry envelopes that failed to send. + * + * `makeOfflineTransport` from `@sentry/core` stores an envelope here when its + * send throws (network error or the request deadline), and the next CLI run + * sends the stored envelopes again (see `offline-transport.ts`). Rows live in + * `telemetry_envelope_queue` as Brotli-compressed serialized envelopes, in a + * BLOB since attachments can be binary. + * + * The store is capped by entry count and total compressed bytes. Over a + * cap, it drops envelopes without an error event first, then errors, oldest + * first. Rows past their max age ({@link STORE_LIMITS}) are deleted on read + * and write. Reads claim rows with `DELETE ... RETURNING`, so two CLI + * processes never send the same envelope. Store failures never reach the + * CLI: each operation logs at debug level and carries on. + */ + +import { + brotliCompressSync, + brotliDecompressSync, + constants as zlib, +} from "node:zlib"; +import { + type Envelope, + envelopeContainsItemType, + type OfflineStore, + parseEnvelope, + serializeEnvelope, + suppressTracing, +} from "@sentry/core"; +import { getDatabase } from "../db/index.js"; +import { logger } from "../logger.js"; + +const log = logger.withTag("telemetry-store"); + +const DAY_MS = 24 * 60 * 60 * 1000; + +/** Rows past their max age, given the limits and the current time. */ +const EXPIRED = "created_at < CASE WHEN is_error THEN ? ELSE ? END"; + +/** Limits on what the store keeps. */ +export type StoreLimits = { + /** Most envelopes kept. */ + maxEntries: number; + /** Most compressed bytes kept across all envelopes. */ + maxBytes: number; + /** Max age, in milliseconds since stored, of an envelope with an error event. */ + maxErrorAgeMs: number; + /** Max age of every other envelope. */ + maxAgeMs: number; +}; + +/** + * Default limits. + * + * Relay drops sessions (`max_session_secs_in_past`) and transactions (the + * metrics aggregator's `max_secs_in_past`) older than 5 days, so envelopes + * without an error event keep 4 days, leaving a day for clock skew. Relay + * accepts an error event for the project's retention (90 days by default) + * and past that only replaces its timestamp, so errors keep 30 days. + * + * Brotli shrinks the CLI's envelopes about 8x, so the byte cap holds about + * 30 MiB of them and only binds for unusually large ones. + */ +export const STORE_LIMITS: StoreLimits = { + maxEntries: 100, + maxBytes: 4 * 1024 * 1024, + maxErrorAgeMs: 30 * DAY_MS, + maxAgeMs: 4 * DAY_MS, +}; + +type StoredRow = { + id: number; + created_at: number; + envelope: Uint8Array; +}; + +/** The store's operations. */ +type EnvelopeStore = { + /** Store an envelope, then drop expired rows and the rows over the caps. */ + push: OfflineStore["push"]; + /** + * Claim up to `limit` of the oldest unexpired envelopes, deleting them from + * the store. Pushing a claimed envelope back keeps its original age. + */ + claim(limit: number): Envelope[]; +}; + +/** Run a store operation without DB spans; log and fall back on failure. */ +function guarded(operation: string, fallback: T, fn: () => T): T { + try { + return suppressTracing(fn); + } catch (error) { + log.debug(`Failed to ${operation}`, error); + return fallback; + } +} + +/** + * Brotli quality for stored envelopes. On envelopes the CLI stored with + * ingest stalled, 6 compressed about 8x (gzip: 7x, zstd level 3: 6.8x) at + * well under a millisecond per envelope; 11 gained 6% for 50x the time. + */ +const BROTLI_QUALITY = 6; + +/** The bytes the store keeps for an envelope. */ +export function compressEnvelope(envelope: Envelope): Uint8Array { + const serialized = serializeEnvelope(envelope); + const bytes = + typeof serialized === "string" ? Buffer.from(serialized) : serialized; + return brotliCompressSync(bytes, { + params: { + [zlib.BROTLI_PARAM_QUALITY]: BROTLI_QUALITY, + [zlib.BROTLI_PARAM_SIZE_HINT]: bytes.byteLength, + }, + }); +} + +/** + * Create the store. + * + * @param limits - Caps and max age; defaults to {@link STORE_LIMITS} + * @param now - Clock, injectable for tests + */ +export function createEnvelopeStore( + limits: StoreLimits = STORE_LIMITS, + now: () => number = Date.now, +): EnvelopeStore { + /** When each envelope claimed by this process was first stored. */ + const storedAt = new WeakMap(); + + function deleteExpired(db: ReturnType): void { + const current = now(); + db.query(`DELETE FROM telemetry_envelope_queue WHERE ${EXPIRED}`).run( + current - limits.maxErrorAgeMs, + current - limits.maxAgeMs, + ); + } + + function insert(envelope: Envelope): void { + const bytes = compressEnvelope(envelope); + // It could never be kept, and inserting it would put every row over + // the byte cap in the trim below, emptying the store. + if (bytes.byteLength > limits.maxBytes) { + log.debug("Dropped unsent telemetry larger than the store's byte cap"); + return; + } + const createdAt = storedAt.get(envelope) ?? now(); + const isError = envelopeContainsItemType(envelope, ["event"]) ? 1 : 0; + const db = getDatabase(); + db.transaction(() => { + db.query( + "INSERT INTO telemetry_envelope_queue (created_at, is_error, size, envelope) VALUES (?, ?, ?, ?)", + ).run(createdAt, isError, bytes.byteLength, bytes); + // Expired rows go first, in their own statement, so they don't count + // against the caps in the trim below. + deleteExpired(db); + // Keep errors first, then the newest: rows past either cap are the + // oldest envelopes without an error, then the oldest errors. + db.query( + `DELETE FROM telemetry_envelope_queue WHERE id IN ( + SELECT id FROM ( + SELECT id, + ROW_NUMBER() OVER kept AS position, + SUM(size) OVER kept AS total + FROM telemetry_envelope_queue + WINDOW kept AS (ORDER BY is_error DESC, created_at DESC, id DESC) + ) WHERE position > ? OR total > ? + )`, + ).run(limits.maxEntries, limits.maxBytes); + })(); + } + + function claim(limit: number): Envelope[] { + return guarded("read stored telemetry", [], () => { + const db = getDatabase(); + const rows = db.transaction(() => { + deleteExpired(db); + return db + .query( + `DELETE FROM telemetry_envelope_queue WHERE id IN ( + SELECT id FROM telemetry_envelope_queue + ORDER BY created_at, id LIMIT ? + ) RETURNING id, created_at, envelope`, + ) + .all(limit) as StoredRow[]; + })(); + rows.sort((a, b) => a.created_at - b.created_at || a.id - b.id); + const envelopes: Envelope[] = []; + for (const row of rows) { + try { + const envelope = parseEnvelope(brotliDecompressSync(row.envelope)); + storedAt.set(envelope, row.created_at); + envelopes.push(envelope); + } catch (error) { + log.debug( + "Dropped a stored telemetry envelope that failed to parse", + error, + ); + } + } + return envelopes; + }); + } + + return { + push(envelope) { + guarded("store unsent telemetry", undefined, () => insert(envelope)); + return Promise.resolve(); + }, + claim, + }; +} + +/** Delete every stored envelope; used when telemetry is turned off. */ +export function clearStoredEnvelopes(): void { + guarded("clear stored telemetry", undefined, () => { + const db = getDatabase(); + // Check first: this runs on every opted-out run, and a write to a + // read-only database prints a warning. + if (db.query("SELECT 1 FROM telemetry_envelope_queue LIMIT 1").get()) { + db.query("DELETE FROM telemetry_envelope_queue").run(); + } + }); +} diff --git a/packages/cli/src/lib/telemetry/offline-transport.ts b/packages/cli/src/lib/telemetry/offline-transport.ts new file mode 100644 index 000000000..fd4229455 --- /dev/null +++ b/packages/cli/src/lib/telemetry/offline-transport.ts @@ -0,0 +1,186 @@ +/** + * Telemetry transport that keeps envelopes it fails to send. + * + * Wraps a transport factory (the zstd transport by default) with + * `makeOfflineTransport` from `@sentry/core`, backed by the SQLite store in + * `offline-store.ts`. The SDK stores an envelope whose send throws, which + * includes the request deadline. It returns any HTTP response as is, so a + * 4xx or 5xx is dropped, and it never stores client reports. Metrics + * (`trace_metric`) aren't stored either: they are sent as usual and dropped + * on failure. A long-running command emits runtime metrics every few + * seconds, and old ones are worth little; stored, they would push sessions + * and transactions out of the store's caps. + * + * On top of that: + * - A per-transport circuit breaker. After the first network error or + * timeout, `shouldSend` returns false, so later envelopes go straight to + * the store with no network wait. With ingest stalled, a short command then + * waits for at most one request deadline on exit instead of one per round + * of sends. After {@link BREAKER_COOLDOWN_MS}, the next envelope is sent as + * a probe: a response closes the breaker and sends a batch of stored + * envelopes, and a failure opens it for another cooldown. Only long-running + * commands (`sentry mcp`, `sentry local`) live long enough to probe. + * - {@link StoringTransport.sendStored}, which each run calls at startup, in + * place of the SDK's own retries. Those run from unref'd timers (5 s + * backoff, 100 ms after a success) that rarely fire before a short command + * exits, and could extend the exit when they do. + */ + +import { + envelopeContainsItemType, + makeOfflineTransport, + type Transport, + type TransportMakeRequestResponse, +} from "@sentry/core"; +import { logger } from "../logger.js"; +import { createEnvelopeStore } from "./offline-store.js"; +import { + makeCompressedTransport, + type NodeTransportOptions, +} from "./zstd-transport.js"; + +const log = logger.withTag("telemetry-store"); + +/** Stored envelopes a run sends at startup, and after the breaker closes. */ +export const STORED_ENVELOPES_PER_RUN = 10; + +/** + * How long the breaker stays open before a probe. + * + * Short commands never live this long, so they keep one deadline and then + * disk. For a long-running process (`sentry mcp`, `sentry local`), a stalled + * ingest costs one deadline-bound request every 5 minutes, and one batch of + * {@link STORED_ENVELOPES_PER_RUN} stored envelopes goes out within 5 minutes + * of ingest or the network coming back. The rest wait for later runs, so + * each recovery adds a bounded amount of work. + */ +export const BREAKER_COOLDOWN_MS = 5 * 60 * 1000; + +/** + * Breaker state. `closed` sends; `open` stores until the cooldown since + * `openedAt` ends; `probing` stores while the one probe is in flight. + */ +type BreakerState = + | { kind: "closed" } + | { kind: "open"; openedAt: number } + | { kind: "probing" }; + +/** A transport that can also send envelopes stored by earlier runs. */ +export type StoringTransport = Transport & { + /** + * Claim up to `limit` stored envelopes and send them in the background. + * One that fails to send again goes back to the store with its original + * age. Does nothing while the breaker is open. + */ + sendStored(limit: number): void; +}; + +/** + * Create the transport. + * + * @param options - Transport options from `Sentry.init` + * @param makeTransport - Inner transport factory; carries the request deadline + * @param now - Clock for the breaker cooldown, injectable for tests + */ +export function makeStoringTransport( + options: NodeTransportOptions, + makeTransport: ( + options: NodeTransportOptions, + ) => Transport = makeCompressedTransport, + now: () => number = Date.now, +): StoringTransport { + const store = createEnvelopeStore(); + let breaker: BreakerState = { kind: "closed" }; + + function shouldSend(): boolean { + if (breaker.kind === "closed") { + return true; + } + if ( + breaker.kind === "open" && + now() - breaker.openedAt >= BREAKER_COOLDOWN_MS + ) { + breaker = { kind: "probing" }; + return true; + } + return false; + } + + function onFailure(reason: unknown): void { + if (breaker.kind !== "open") { + log.debug("Telemetry ingest unreachable; storing envelopes", reason); + } + breaker = { kind: "open", openedAt: now() }; + } + + function onResponse(): void { + // Only the probe's answer closes the breaker. A request that started + // while it was closed and answers after another one failed (say, in a + // parallel drain batch) leaves it open, so one run never drains more + // than one batch per recovery. + if (breaker.kind !== "probing") { + return; + } + breaker = { kind: "closed" }; + log.debug("Telemetry ingest reachable again; sending stored envelopes"); + sendStored(STORED_ENVELOPES_PER_RUN); + } + + function sendStored(limit: number): void { + if (breaker.kind !== "closed") { + return; + } + for (const envelope of store.claim(limit)) { + // Relay corrects item timestamps by the gap between `sent_at` and + // when it received the envelope, so a stale `sent_at` would shift + // them forward. The SDK's own retries reset it the same way. + envelope[0].sent_at = new Date().toISOString(); + transport.send(envelope).then(undefined, (error: unknown) => { + log.debug("Failed to resend stored telemetry", error); + }); + } + } + + const transport = makeOfflineTransport( + (innerOptions: NodeTransportOptions) => { + const inner = makeTransport(innerOptions); + return { + send: async (envelope): Promise => { + let response: TransportMakeRequestResponse; + try { + response = await inner.send(envelope); + } catch (error) { + onFailure(error); + throw error; + } + // Any response, even a 4xx or 5xx, means ingest is reachable. The + // SDK records no outcome for a dropped 5xx, and the CLI's client + // reports show network errors, not 5xx, as the real loss. + onResponse(); + return response; + }, + flush: (timeout) => inner.flush(timeout), + }; + }, + )({ + ...options, + createStore: () => ({ + push: store.push, + unshift: store.push, + // Stored envelopes are sent only by sendStored. The SDK's timers shift + // one more after each success (and on a backoff after each failure); + // one that fires during the exit flush starts a request that keeps the + // process alive past it, and those can chain while others are in flight. + shift: () => Promise.resolve(undefined), + }), + shouldSend, + shouldStore: (envelope) => + !envelopeContainsItemType(envelope, ["trace_metric"]), + }); + + return { + send: (envelope) => transport.send(envelope), + flush: (timeout) => transport.flush(timeout), + sendStored, + }; +} diff --git a/packages/cli/src/lib/telemetry/zstd-transport.ts b/packages/cli/src/lib/telemetry/zstd-transport.ts index 43d46852c..7e97e6546 100644 --- a/packages/cli/src/lib/telemetry/zstd-transport.ts +++ b/packages/cli/src/lib/telemetry/zstd-transport.ts @@ -48,7 +48,7 @@ import { import { makeNodeTransport } from "@sentry/node"; /** `@sentry/node` v11 no longer re-exports this type from its entry point. */ -type NodeTransportOptions = Parameters[0]; +export type NodeTransportOptions = Parameters[0]; type HttpModule = NonNullable; diff --git a/packages/cli/test/lib/db/concurrent-worker.ts b/packages/cli/test/lib/db/concurrent-worker.ts index 3164e0ac1..7e93cc2d7 100644 --- a/packages/cli/test/lib/db/concurrent-worker.ts +++ b/packages/cli/test/lib/db/concurrent-worker.ts @@ -9,6 +9,7 @@ * write-dsn - Write a unique DSN cache entry * write-project - Write a unique project cache entry * read-write - Mixed read/write operations + * claim-envelopes - Claim stored telemetry envelopes until none are left */ import { mkdirSync } from "node:fs"; @@ -37,6 +38,8 @@ const { setCachedDsn, getCachedDsn } = const { setCachedProject, getCachedProject } = await import("../../../src/lib/db/project-cache.js"); const { closeDatabase } = await import("../../../src/lib/db/index.js"); +const { createEnvelopeStore } = + await import("../../../src/lib/telemetry/offline-store.js"); type WorkerResult = { workerId: string; @@ -142,6 +145,24 @@ async function readWrite(): Promise { }; } +function claimEnvelopes(): WorkerResult { + const store = createEnvelopeStore(); + const claimed: unknown[] = []; + for (;;) { + const batch = store.claim(3); + if (batch.length === 0) { + break; + } + claimed.push(...batch.map((envelope) => envelope[0].event_id)); + } + return { + workerId, + operation: "claim-envelopes", + success: true, + data: claimed, + }; +} + async function main(): Promise { let result: WorkerResult; @@ -156,6 +177,9 @@ async function main(): Promise { case "read-write": result = await readWrite(); break; + case "claim-envelopes": + result = claimEnvelopes(); + break; default: result = { workerId, diff --git a/packages/cli/test/lib/db/concurrent.test.ts b/packages/cli/test/lib/db/concurrent.test.ts index 1ed19ed18..9caa16e55 100644 --- a/packages/cli/test/lib/db/concurrent.test.ts +++ b/packages/cli/test/lib/db/concurrent.test.ts @@ -10,6 +10,7 @@ import { spawn } from "node:child_process"; import { join } from "node:path"; +import { createEnvelope } from "@sentry/core"; import { beforeEach, describe, expect, test } from "vitest"; function noop(): void { @@ -22,6 +23,7 @@ import { closeDatabase, } from "../../../src/lib/db/index.js"; import { getCachedProject } from "../../../src/lib/db/project-cache.js"; +import { createEnvelopeStore } from "../../../src/lib/telemetry/offline-store.js"; import { useTestConfigDir } from "../../helpers.js"; const WORKER_SCRIPT = join(import.meta.dirname, "concurrent-worker.ts"); @@ -195,6 +197,29 @@ describe("concurrent database access", () => { } }); + test("processes claiming stored telemetry never get the same envelope", async () => { + const store = createEnvelopeStore(); + const ids = Array.from({ length: 60 }, (_, i) => `event-${i}`); + for (const id of ids) { + await store.push( + createEnvelope({ event_id: id }, [ + [{ type: "event" }, { event_id: id }], + ] as never), + ); + } + closeDatabase(); + + const results = await spawnWorkersConcurrently( + getConfigDir(), + 4, + "claim-envelopes", + ); + + expect(results.filter((r) => !r.success)).toHaveLength(0); + const claimed = results.flatMap((r) => r.data as string[]); + expect(claimed.sort()).toEqual([...ids].sort()); + }); + test("handles contention without SQLITE_BUSY errors", async () => { // Run a larger batch to increase contention likelihood const workerCount = 10; diff --git a/packages/cli/test/lib/db/schema.test.ts b/packages/cli/test/lib/db/schema.test.ts index d7d42f71c..e328c5533 100644 --- a/packages/cli/test/lib/db/schema.test.ts +++ b/packages/cli/test/lib/db/schema.test.ts @@ -260,6 +260,7 @@ describe("TABLE_SCHEMAS persistence", () => { "project_cache", "project_root_cache", "repo_cache", + "telemetry_envelope_queue", ]); }); }); diff --git a/packages/cli/test/lib/mcp/telemetry.test.ts b/packages/cli/test/lib/mcp/telemetry.test.ts index 368fae2ea..5f065a5f7 100644 --- a/packages/cli/test/lib/mcp/telemetry.test.ts +++ b/packages/cli/test/lib/mcp/telemetry.test.ts @@ -1,8 +1,15 @@ +import { createEnvelope, type Envelope } from "@sentry/core"; import type { NodeOptions } from "@sentry/node"; import { afterEach, beforeEach, describe, expect, test, vi } from "vitest"; import { CLI_VERSION, SENTRY_CLI_DSN } from "../../../src/lib/constants.js"; import type { McpServerConfig } from "../../../src/lib/mcp/config.js"; import { useEnvSandbox, useTestConfigDir } from "../../helpers.js"; +import { + closeIngestServers, + noop, + startMockIngest, + startRefusedIngest, +} from "../telemetry/ingest-servers.js"; const sentry = vi.hoisted(() => ({ init: vi.fn((_options: NodeOptions) => undefined), @@ -47,11 +54,52 @@ beforeEach(() => { sentry.init.mockClear(); }); -afterEach(() => { +afterEach(async () => { vi.unstubAllGlobals(); vi.resetModules(); + await closeIngestServers(); }); +function eventEnvelope(id: string): Envelope { + return createEnvelope({ event_id: id }, [ + [{ type: "event" }, { event_id: id, message: "hi" }], + ] as never); +} + +/** + * The database and store modules the server will load (modules are reset + * after each test), with the database already open. Opening it with + * telemetry on `require`s the telemetry module, which fails under vitest. + */ +async function openStore() { + const { getDatabase } = await import("../../../src/lib/db/index.js"); + const { createEnvelopeStore } = + await import("../../../src/lib/telemetry/offline-store.js"); + const optedOut = process.env.SENTRY_CLI_NO_TELEMETRY; + process.env.SENTRY_CLI_NO_TELEMETRY = "1"; + getDatabase(); + if (optedOut === undefined) { + delete process.env.SENTRY_CLI_NO_TELEMETRY; + } + return { + store: createEnvelopeStore(), + storedCount: () => + ( + getDatabase() + .query("SELECT COUNT(*) AS count FROM telemetry_envelope_queue") + .get() as { count: number } + ).count, + }; +} + +/** The transport the server's `Sentry.init` options build for `url`. */ +function buildTransport(options: NodeOptions, url: string) { + if (!options.transport) { + throw new Error("Sentry.init got no transport"); + } + return options.transport({ url, recordDroppedEvent: noop } as never); +} + describe("initMcpTelemetry", () => { test("reports to the CLI's project", async () => { const options = await initOptions(); @@ -128,4 +176,41 @@ describe("initMcpTelemetry", () => { cwd.mockRestore(); } }); + + test("stores telemetry that fails to send", async () => { + const { storedCount } = await openStore(); + const { url } = await startRefusedIngest(); + const transport = buildTransport(await initOptions(), url); + + await expect(transport.send(eventEnvelope("a"))).resolves.toEqual({}); + + expect(storedCount()).toBe(1); + }); + + test("sends what earlier runs stored at startup", async () => { + const { store, storedCount } = await openStore(); + await store.push(eventEnvelope("stored")); + const ingest = await startMockIngest(() => ({ statusCode: 200 })); + // The real `init` builds the transport; the mock does the same so the + // server's startup drain has one. + sentry.init.mockImplementationOnce((options) => { + buildTransport(options, ingest.url); + }); + + await initOptions(); + + await vi.waitFor(() => expect(ingest.captures).toHaveLength(1)); + expect(storedCount()).toBe(0); + }); + + test("opting out clears stored telemetry", async () => { + const { store, storedCount } = await openStore(); + await store.push(eventEnvelope("stored")); + expect(storedCount()).toBe(1); + process.env.SENTRY_CLI_NO_TELEMETRY = "1"; + + await initOptions(); + + expect(storedCount()).toBe(0); + }); }); diff --git a/packages/cli/test/lib/telemetry-offline.test.ts b/packages/cli/test/lib/telemetry-offline.test.ts new file mode 100644 index 000000000..8d85a40a3 --- /dev/null +++ b/packages/cli/test/lib/telemetry-offline.test.ts @@ -0,0 +1,101 @@ +/** + * How `withTelemetry` uses the store of telemetry that failed to send: + * it resends stored envelopes when telemetry is on, clears them when the + * user opted out, and leaves them alone in library mode. + */ + +import { createEnvelope, type Envelope } from "@sentry/core"; +import { afterEach, beforeEach, describe, expect, test, vi } from "vitest"; +import { getDatabase } from "../../src/lib/db/index.js"; +import { initSentry, withTelemetry } from "../../src/lib/telemetry.js"; +import { createEnvelopeStore } from "../../src/lib/telemetry/offline-store.js"; +import { useTestConfigDir } from "../helpers.js"; + +const sent = vi.hoisted(() => [] as Envelope[]); + +// Record envelopes instead of sending them to the production DSN. +vi.mock("../../src/lib/telemetry/zstd-transport.js", async (importOriginal) => { + const actual = + await importOriginal< + typeof import("../../src/lib/telemetry/zstd-transport.js") + >(); + return { + ...actual, + makeCompressedTransport: () => ({ + send: (envelope: Envelope) => { + sent.push(envelope); + return Promise.resolve({ statusCode: 200 }); + }, + flush: () => Promise.resolve(true), + }), + }; +}); + +const NO_TELEMETRY = "SENTRY_CLI_NO_TELEMETRY"; + +function storedCount(): number { + return ( + getDatabase() + .query("SELECT COUNT(*) AS count FROM telemetry_envelope_queue") + .get() as { count: number } + ).count; +} + +async function storeEnvelope(id: string): Promise { + await createEnvelopeStore().push( + createEnvelope({ event_id: id }, [ + [{ type: "event" }, { event_id: id }], + ] as never), + ); +} + +describe("withTelemetry and stored telemetry", () => { + useTestConfigDir("telemetry-offline-run-"); + let savedNoTelemetry: string | undefined; + + beforeEach(async () => { + savedNoTelemetry = process.env[NO_TELEMETRY]; + sent.length = 0; + await storeEnvelope("stored"); + }); + + afterEach(() => { + if (savedNoTelemetry === undefined) { + delete process.env[NO_TELEMETRY]; + } else { + process.env[NO_TELEMETRY] = savedNoTelemetry; + } + initSentry(false); + }); + + test("sends stored envelopes when telemetry is on", async () => { + delete process.env[NO_TELEMETRY]; + + await withTelemetry(() => undefined); + + expect(sent.map((envelope) => envelope[0].event_id)).toContain("stored"); + expect(storedCount()).toBe(0); + }); + + test("clears stored envelopes when telemetry is off", async () => { + process.env[NO_TELEMETRY] = "1"; + + await withTelemetry(() => undefined); + + expect(storedCount()).toBe(0); + expect(sent).toEqual([]); + }); + + test("library mode leaves the store alone", async () => { + process.env[NO_TELEMETRY] = "1"; + await withTelemetry(() => undefined, { libraryMode: true }); + + delete process.env[NO_TELEMETRY]; + await withTelemetry(() => undefined, { libraryMode: true }); + + expect(storedCount()).toBe(1); + expect(sent.map((envelope) => envelope[0].event_id)).not.toContain( + "stored", + ); + }); +}); diff --git a/packages/cli/test/lib/telemetry/ingest-servers.ts b/packages/cli/test/lib/telemetry/ingest-servers.ts new file mode 100644 index 000000000..68098af8e --- /dev/null +++ b/packages/cli/test/lib/telemetry/ingest-servers.ts @@ -0,0 +1,150 @@ +/** + * Local stand-ins for the Sentry ingest endpoint, shared by the telemetry + * transport e2e tests: one that answers, one that stalls after connecting, + * one that refuses connections, and a proxy that stalls after CONNECT. + * + * Every server started here is tracked; call {@link closeIngestServers} in + * `afterEach` so a test never leaks a listening socket into the next one. + */ + +import { createServer, type IncomingMessage } from "node:http"; +import { + type AddressInfo, + createServer as createTcpServer, + type Socket, +} from "node:net"; + +/** No-op for SDK callbacks that require a function but return nothing meaningful. */ +export function noop(): void { + // intentionally empty +} + +/** One request a mock ingest received. */ +export type CapturedRequest = { + headers: IncomingMessage["headers"]; + body: Buffer; +}; + +/** What these helpers need from an HTTP or TCP server. */ +type LocalServer = { + listen(port: number, host: string, callback: () => void): unknown; + address(): unknown; + close(callback: () => void): unknown; +}; + +const startedServers: LocalServer[] = []; + +/** Close every server started since the last call, in parallel. */ +export async function closeIngestServers(): Promise { + await Promise.all( + startedServers.splice(0).map( + (s) => + new Promise((resolve) => { + s.close(() => resolve()); + }), + ), + ); +} + +function listen(server: LocalServer, track = true): Promise { + if (track) { + startedServers.push(server); + } + return new Promise((resolve) => { + server.listen(0, "127.0.0.1", () => { + resolve((server.address() as AddressInfo).port); + }); + }); +} + +function envelopeUrl(port: number): string { + return `http://127.0.0.1:${port}/api/0/envelope/`; +} + +/** + * Start a server that accepts connections and never answers, like an + * ingest endpoint behind a stalled network path. `closed` resolves when + * the client gives up and its socket closes; `connections()` counts the + * connections it accepted. + */ +export async function startStalledIngest(): Promise<{ + url: string; + closed: Promise; + connections: () => number; +}> { + let onClosed: () => void = noop; + const closed = new Promise((resolve) => { + onClosed = resolve; + }); + let connections = 0; + const server = createTcpServer((socket: Socket) => { + connections += 1; + socket.on("error", noop); + socket.on("close", () => onClosed()); + // Read (and drop) the request so the client's hang-up is seen. + socket.resume(); + }); + const port = await listen(server); + return { url: envelopeUrl(port), closed, connections: () => connections }; +} + +/** An ingest URL on a port nothing listens on, so connections are refused. */ +export async function startRefusedIngest(): Promise<{ url: string }> { + const server = createTcpServer(); + const port = await listen(server, false); + await new Promise((resolve) => { + server.close(() => resolve()); + }); + return { url: envelopeUrl(port) }; +} + +/** + * Start an HTTP proxy that accepts the CONNECT tunnel and then goes + * silent. Routes the transport through the SDK's `makeNodeTransport`. + */ +export async function startStalledProxy(): Promise<{ + url: string; + closed: Promise; +}> { + let onClosed: () => void = noop; + const closed = new Promise((resolve) => { + onClosed = resolve; + }); + const server = createServer(); + server.on("connect", (_req, socket) => { + socket.on("error", noop); + socket.on("close", () => onClosed()); + // HTTP servers allow half-open sockets: close ours when the client + // hangs up, so `closed` proves the client released its socket. + socket.on("end", () => socket.end()); + socket.resume(); + socket.write("HTTP/1.1 200 Connection Established\r\n\r\n"); + }); + const port = await listen(server); + return { url: `http://127.0.0.1:${port}`, closed }; +} + +/** Start an ingest that records each request and answers with `responder`. */ +export async function startMockIngest( + responder: (req: IncomingMessage) => { + statusCode: number; + headers?: Record; + }, +): Promise<{ + url: string; + captures: CapturedRequest[]; +}> { + const captures: CapturedRequest[] = []; + const server = createServer((req, res) => { + const chunks: Buffer[] = []; + req.on("data", (c: Buffer) => chunks.push(c)); + req.on("end", () => { + captures.push({ headers: req.headers, body: Buffer.concat(chunks) }); + const response = responder(req); + res.writeHead(response.statusCode, response.headers ?? {}); + res.end(); + }); + }); + const port = await listen(server); + return { url: envelopeUrl(port), captures }; +} diff --git a/packages/cli/test/lib/telemetry/offline-store.test.ts b/packages/cli/test/lib/telemetry/offline-store.test.ts new file mode 100644 index 000000000..2b53c3643 --- /dev/null +++ b/packages/cli/test/lib/telemetry/offline-store.test.ts @@ -0,0 +1,239 @@ +/** + * Tests for the SQLite store of telemetry envelopes that failed to send. + * Claims from several processes at once are covered in + * test/lib/db/concurrent.test.ts. + */ + +import { randomBytes } from "node:crypto"; +import { writeFileSync } from "node:fs"; +import { join } from "node:path"; +import { createEnvelope, type Envelope } from "@sentry/core"; +import { afterEach, beforeEach, describe, expect, test } from "vitest"; +import { + CONFIG_DIR_ENV_VAR, + closeDatabase, + getDatabase, +} from "../../../src/lib/db/index.js"; +import { __setDriverForTests } from "../../../src/lib/db/sqlite.js"; +import { + clearStoredEnvelopes, + compressEnvelope, + createEnvelopeStore, + STORE_LIMITS, + type StoreLimits, +} from "../../../src/lib/telemetry/offline-store.js"; +import { useTestConfigDir } from "../../helpers.js"; + +const DAY_MS = 24 * 60 * 60 * 1000; + +function eventEnvelope(id: string, message = "hi"): Envelope { + return createEnvelope({ event_id: id, sent_at: "2026-01-01T00:00:00Z" }, [ + [{ type: "event" }, { event_id: id, message }], + ] as never); +} + +/** An envelope without an error event; the header ID only labels it. */ +function transactionEnvelope(id: string): Envelope { + return createEnvelope({ event_id: id }, [ + [{ type: "transaction" }, { event_id: id, type: "transaction" }], + ] as never); +} + +function eventIds(envelopes: Envelope[]): unknown[] { + return envelopes.map((envelope) => envelope[0].event_id); +} + +function storedCount(): number { + return ( + getDatabase() + .query("SELECT COUNT(*) AS count FROM telemetry_envelope_queue") + .get() as { count: number } + ).count; +} + +/** A store on a settable clock. */ +function storeAt(limits: StoreLimits = STORE_LIMITS) { + const clock = { now: 1_000_000 }; + return { clock, store: createEnvelopeStore(limits, () => clock.now) }; +} + +describe.each(["node", "wasm"] as const)( + "createEnvelopeStore [%s driver]", + (kind) => { + useTestConfigDir("telemetry-store-"); + + beforeEach(() => { + __setDriverForTests(kind); + }); + + afterEach(() => { + closeDatabase(); + __setDriverForTests(null); + }); + + test("round-trips JSON and binary items", async () => { + const { store } = storeAt(); + const bytes = new Uint8Array([0, 1, 2, 255, 10, 13]); + const envelope = createEnvelope({ event_id: "a" }, [ + [{ type: "event" }, { event_id: "a", message: "é ✓" }], + [ + { type: "attachment", length: bytes.length, filename: "f.bin" }, + bytes, + ], + ] as never); + + await store.push(envelope); + const [claimed] = store.claim(10); + + expect(claimed?.[0]).toEqual(envelope[0]); + expect(claimed?.[1][0]).toEqual(envelope[1][0]); + expect(claimed?.[1][1]?.[0]).toEqual(envelope[1][1]?.[0]); + expect(Buffer.from(claimed?.[1][1]?.[1] as Uint8Array)).toEqual( + Buffer.from(bytes), + ); + }); + + test("claims the oldest first and deletes what it claims", async () => { + const { clock, store } = storeAt(); + for (const id of ["a", "b", "c"]) { + await store.push(eventEnvelope(id)); + clock.now += 1; + } + + expect(eventIds(store.claim(2))).toEqual(["a", "b"]); + expect(storedCount()).toBe(1); + expect(eventIds(store.claim(2))).toEqual(["c"]); + expect(store.claim(2)).toEqual([]); + }); + + test("keeps only the newest maxEntries envelopes", async () => { + const { clock, store } = storeAt({ ...STORE_LIMITS, maxEntries: 3 }); + for (const id of ["a", "b", "c", "d", "e"]) { + await store.push(transactionEnvelope(id)); + clock.now += 1; + } + + expect(eventIds(store.claim(10))).toEqual(["c", "d", "e"]); + }); + + test("keeps the newest envelopes that fit in maxBytes", async () => { + // The cap counts stored, compressed bytes. + const size = compressEnvelope(eventEnvelope("a")).byteLength; + const { clock, store } = storeAt({ ...STORE_LIMITS, maxBytes: size * 2 }); + for (const id of ["a", "b", "c"]) { + await store.push(eventEnvelope(id)); + clock.now += 1; + } + expect(eventIds(store.claim(10))).toEqual(["b", "c"]); + + // One envelope bigger than the cap is not kept, and doesn't push out + // the envelopes that fit. + await store.push(eventEnvelope("d")); + await store.push( + eventEnvelope("big", randomBytes(size * 4).toString("hex")), + ); + expect(eventIds(store.claim(10))).toEqual(["d"]); + }); + + test("over a cap, drops envelopes without an error before errors", async () => { + const { clock, store } = storeAt({ ...STORE_LIMITS, maxEntries: 3 }); + await store.push(eventEnvelope("e1")); + clock.now += 1; + for (const id of ["t1", "t2"]) { + await store.push(transactionEnvelope(id)); + clock.now += 1; + } + await store.push(eventEnvelope("e2")); + clock.now += 1; + await store.push(transactionEnvelope("t3")); + + expect(eventIds(store.claim(10))).toEqual(["e1", "e2", "t3"]); + }); + + test("expired rows don't count against the caps", async () => { + const { clock, store } = storeAt({ ...STORE_LIMITS, maxEntries: 2 }); + for (const id of ["e1", "e2"]) { + await store.push(eventEnvelope(id)); + } + clock.now += STORE_LIMITS.maxErrorAgeMs + 1; + // The expired errors would rank first and push this out of the cap. + await store.push(transactionEnvelope("t1")); + + expect(eventIds(store.claim(10))).toEqual(["t1"]); + }); + + test("deletes envelopes without an error event after maxAgeMs", async () => { + const { clock, store } = storeAt(); + await store.push(transactionEnvelope("old")); + clock.now += STORE_LIMITS.maxAgeMs - 1; + await store.push(transactionEnvelope("new")); + clock.now += 2; + + expect(eventIds(store.claim(10))).toEqual(["new"]); + expect(storedCount()).toBe(0); + }); + + test("keeps errors until maxErrorAgeMs", async () => { + const { clock, store } = storeAt(); + await store.push(eventEnvelope("error")); + await store.push(transactionEnvelope("transaction")); + clock.now += STORE_LIMITS.maxAgeMs + 1; + expect(eventIds(store.claim(10))).toEqual(["error"]); + + await store.push(eventEnvelope("old error")); + clock.now += STORE_LIMITS.maxErrorAgeMs + 1; + expect(store.claim(10)).toEqual([]); + expect(storedCount()).toBe(0); + }); + + test("a claimed envelope pushed back keeps its original age", async () => { + const { clock, store } = storeAt(); + await store.push(transactionEnvelope("a")); + clock.now += 3 * DAY_MS; + const [claimed] = store.claim(1); + await store.push(claimed as Envelope); + clock.now += 2 * DAY_MS; + + expect(store.claim(10)).toEqual([]); + }); + + test("drops a row that no longer parses", async () => { + const { store } = storeAt(); + getDatabase() + .query( + "INSERT INTO telemetry_envelope_queue (created_at, size, envelope) VALUES (?, 3, ?)", + ) + .run(1_000_000, Buffer.from("{{{")); + await store.push(eventEnvelope("a")); + + expect(eventIds(store.claim(10))).toEqual(["a"]); + expect(storedCount()).toBe(0); + }); + + test("never throws when the database is unavailable", async () => { + const { store } = storeAt(); + const blocker = join(process.env[CONFIG_DIR_ENV_VAR] as string, "file"); + writeFileSync(blocker, ""); + closeDatabase(); + process.env[CONFIG_DIR_ENV_VAR] = join(blocker, "nested"); + + await expect(store.push(eventEnvelope("a"))).resolves.toBeUndefined(); + expect(store.claim(10)).toEqual([]); + expect(() => clearStoredEnvelopes()).not.toThrow(); + }); + }, +); + +describe("clearStoredEnvelopes", () => { + useTestConfigDir("telemetry-store-clear-"); + + test("deletes every stored envelope", async () => { + const { store } = storeAt(); + await store.push(eventEnvelope("a")); + await store.push(eventEnvelope("b")); + + clearStoredEnvelopes(); + + expect(storedCount()).toBe(0); + }); +}); diff --git a/packages/cli/test/lib/telemetry/offline-transport.e2e.test.ts b/packages/cli/test/lib/telemetry/offline-transport.e2e.test.ts new file mode 100644 index 000000000..2e8f38f6f --- /dev/null +++ b/packages/cli/test/lib/telemetry/offline-transport.e2e.test.ts @@ -0,0 +1,196 @@ +/** + * E2E tests for the transport that stores telemetry it fails to send. + * + * Runs the real zstd transport (with a short request deadline) and the real + * SQLite store against local ingest stand-ins: one that refuses connections, + * one that stalls after connecting, and one that answers. Each "run" builds a + * fresh transport, like a new CLI process. + */ + +import { brotliDecompressSync } from "node:zlib"; +import { createEnvelope, type Envelope, parseEnvelope } from "@sentry/core"; +import { afterEach, describe, expect, test } from "vitest"; +import { getDatabase } from "../../../src/lib/db/index.js"; +import { makeStoringTransport } from "../../../src/lib/telemetry/offline-transport.js"; +import { makeCompressedTransport } from "../../../src/lib/telemetry/zstd-transport.js"; +import { useTestConfigDir } from "../../helpers.js"; +import { + closeIngestServers, + noop, + startMockIngest, + startRefusedIngest, + startStalledIngest, +} from "./ingest-servers.js"; + +const DEADLINE_MS = 300; + +function eventEnvelope(id: string): Envelope { + return createEnvelope({ event_id: id, sent_at: "2026-01-01T00:00:00Z" }, [ + [{ type: "event" }, { event_id: id, message: "hi" }], + ] as never); +} + +/** A transport for one CLI run, with the request deadline shortened. */ +function runTransport(url: string) { + return makeStoringTransport({ url, recordDroppedEvent: noop }, (options) => + makeCompressedTransport(options, DEADLINE_MS), + ); +} + +function storedEventIds(): string[] { + const rows = getDatabase() + .query( + "SELECT envelope FROM telemetry_envelope_queue ORDER BY created_at, id", + ) + .all() as { envelope: Uint8Array }[]; + return rows.map( + (row) => + parseEnvelope(brotliDecompressSync(row.envelope))[0].event_id as string, + ); +} + +/** Event IDs in the bodies a mock ingest received (bodies are small, so uncompressed). */ +function deliveredEventIds(captures: { body: Buffer }[]): string[] { + return captures.map( + (capture) => + JSON.parse(capture.body.toString().split("\n")[0] as string).event_id, + ); +} + +describe("makeStoringTransport (e2e)", () => { + useTestConfigDir("telemetry-offline-"); + afterEach(closeIngestServers); + + test("stores an envelope when ingest refuses the connection", async () => { + const { url } = await startRefusedIngest(); + const transport = runTransport(url); + + await expect(transport.send(eventEnvelope("a"))).resolves.toEqual({}); + + expect(storedEventIds()).toEqual(["a"]); + }); + + test("stores an envelope that times out, then skips the network", async () => { + const ingest = await startStalledIngest(); + const transport = runTransport(ingest.url); + + const started = performance.now(); + await transport.send(eventEnvelope("a")); + expect(performance.now() - started).toBeGreaterThanOrEqual( + DEADLINE_MS - 50, + ); + expect(ingest.connections()).toBe(1); + + const later = performance.now(); + await transport.send(eventEnvelope("b")); + await transport.send(eventEnvelope("c")); + expect(performance.now() - later).toBeLessThan(DEADLINE_MS / 2); + expect(ingest.connections()).toBe(1); + + expect(storedEventIds()).toEqual(["a", "b", "c"]); + }); + + test("sends in flight before the first timeout all finish within one deadline", async () => { + const ingest = await startStalledIngest(); + const transport = runTransport(ingest.url); + + const started = performance.now(); + const first = transport.send(eventEnvelope("a")); + const second = transport.send(eventEnvelope("b")); + await Promise.all([first, second]); + await transport.send(eventEnvelope("c")); + await transport.flush(5000); + + expect(performance.now() - started).toBeLessThan(DEADLINE_MS * 1.8); + expect(storedEventIds().sort()).toEqual(["a", "b", "c"]); + }); + + test("a later run with ingest healthy delivers what an earlier run stored", async () => { + const down = await startRefusedIngest(); + const firstRun = runTransport(down.url); + await firstRun.send(eventEnvelope("a")); + await firstRun.send(eventEnvelope("b")); + expect(storedEventIds()).toEqual(["a", "b"]); + + const ingest = await startMockIngest(() => ({ statusCode: 200 })); + const secondRun = runTransport(ingest.url); + secondRun.sendStored(10); + await secondRun.flush(5000); + + expect(deliveredEventIds(ingest.captures).sort()).toEqual(["a", "b"]); + expect(storedEventIds()).toEqual([]); + const header = JSON.parse( + ingest.captures[0]?.body.toString().split("\n")[0] as string, + ); + expect(Date.parse(header.sent_at)).toBeGreaterThan(Date.now() - 60_000); + }); + + test("a run sends at most the requested number of stored envelopes", async () => { + const down = await startRefusedIngest(); + const firstRun = runTransport(down.url); + for (const id of ["a", "b", "c"]) { + await firstRun.send(eventEnvelope(id)); + } + + const ingest = await startMockIngest(() => ({ statusCode: 200 })); + const secondRun = runTransport(ingest.url); + secondRun.sendStored(2); + await secondRun.flush(5000); + + expect(deliveredEventIds(ingest.captures).sort()).toEqual(["a", "b"]); + expect(storedEventIds()).toEqual(["c"]); + }); + + test("stored envelopes that fail again go back to the store", async () => { + const down = await startRefusedIngest(); + await runTransport(down.url).send(eventEnvelope("a")); + + const stillDown = runTransport(down.url); + stillDown.sendStored(10); + await stillDown.flush(5000); + // The failed resend stores the envelope after the request settles. + await new Promise((resolve) => setImmediate(resolve)); + + expect(storedEventIds()).toEqual(["a"]); + }); + + test("drops a stored envelope that ingest rejects with a 4xx", async () => { + const down = await startRefusedIngest(); + await runTransport(down.url).send(eventEnvelope("a")); + + const ingest = await startMockIngest(() => ({ statusCode: 400 })); + const run = runTransport(ingest.url); + run.sendStored(10); + await run.flush(5000); + + expect(ingest.captures).toHaveLength(1); + expect(storedEventIds()).toEqual([]); + }); + + test("drops an envelope that ingest answers with a 5xx and keeps sending", async () => { + const ingest = await startMockIngest(() => ({ statusCode: 503 })); + const transport = runTransport(ingest.url); + + await transport.send(eventEnvelope("a")); + await transport.send(eventEnvelope("b")); + + expect(deliveredEventIds(ingest.captures)).toEqual(["a", "b"]); + expect(storedEventIds()).toEqual([]); + }); + + test("never stores client reports", async () => { + const { url } = await startRefusedIngest(); + const transport = runTransport(url); + const report = createEnvelope({}, [ + [{ type: "client_report" }, { timestamp: 0, discarded_events: [] }], + ] as never); + + await expect(transport.send(report)).rejects.toThrow(); + + expect( + getDatabase() + .query("SELECT COUNT(*) AS count FROM telemetry_envelope_queue") + .get(), + ).toEqual({ count: 0 }); + }); +}); diff --git a/packages/cli/test/lib/telemetry/offline-transport.test.ts b/packages/cli/test/lib/telemetry/offline-transport.test.ts new file mode 100644 index 000000000..d327d9c33 --- /dev/null +++ b/packages/cli/test/lib/telemetry/offline-transport.test.ts @@ -0,0 +1,209 @@ +/** + * Tests for the circuit breaker of the transport that stores telemetry it + * fails to send. The inner transport is a stand-in that fails or answers on + * demand, and the breaker runs on a settable clock. Wire-level behavior is + * covered in offline-transport.e2e.test.ts. + */ + +import { brotliDecompressSync } from "node:zlib"; +import { createEnvelope, type Envelope, parseEnvelope } from "@sentry/core"; +import { describe, expect, test, vi } from "vitest"; +import { getDatabase } from "../../../src/lib/db/index.js"; +import { createEnvelopeStore } from "../../../src/lib/telemetry/offline-store.js"; +import { + BREAKER_COOLDOWN_MS, + makeStoringTransport, +} from "../../../src/lib/telemetry/offline-transport.js"; +import { useTestConfigDir } from "../../helpers.js"; +import { noop } from "./ingest-servers.js"; + +function eventEnvelope(id: string): Envelope { + return createEnvelope({ event_id: id }, [ + [{ type: "event" }, { event_id: id, message: "hi" }], + ] as never); +} + +function storedEventIds(): unknown[] { + const rows = getDatabase() + .query( + "SELECT envelope FROM telemetry_envelope_queue ORDER BY created_at, id", + ) + .all() as { envelope: Uint8Array }[]; + return rows.map( + (row) => parseEnvelope(brotliDecompressSync(row.envelope))[0].event_id, + ); +} + +/** + * A storing transport over a stand-in ingest. `ingest.up` decides whether a + * send answers 200 or throws; `ingest.sent` lists every envelope it saw. + */ +function setup() { + const clock = { now: 1_000_000 }; + const ingest = { up: false, sent: [] as unknown[] }; + const transport = makeStoringTransport( + { url: "http://127.0.0.1:9/api/0/envelope/", recordDroppedEvent: noop }, + () => ({ + send: async (envelope) => { + ingest.sent.push(envelope[0].event_id); + if (!ingest.up) { + throw new Error("connect ECONNREFUSED"); + } + return { statusCode: 200 }; + }, + flush: () => Promise.resolve(true), + }), + () => clock.now, + ); + return { clock, ingest, transport }; +} + +describe("makeStoringTransport breaker", () => { + useTestConfigDir("telemetry-breaker-"); + + test("probes after the cooldown, re-opens on failure, and drains on recovery", async () => { + const { clock, ingest, transport } = setup(); + + // The first failure opens the breaker; later envelopes skip the network. + await transport.send(eventEnvelope("a")); + await transport.send(eventEnvelope("b")); + clock.now += BREAKER_COOLDOWN_MS - 1; + await transport.send(eventEnvelope("c")); + expect(ingest.sent).toEqual(["a"]); + expect(storedEventIds()).toEqual(["a", "b", "c"]); + + // After the cooldown, one envelope goes out as a probe. It fails, so the + // breaker opens again for a full cooldown. + clock.now += 1; + await transport.send(eventEnvelope("d")); + expect(ingest.sent).toEqual(["a", "d"]); + clock.now += BREAKER_COOLDOWN_MS - 1; + await transport.send(eventEnvelope("e")); + expect(ingest.sent).toEqual(["a", "d"]); + expect(storedEventIds()).toEqual(["a", "b", "c", "d", "e"]); + + // Ingest is back: the next probe succeeds, closes the breaker, and sends + // the stored envelopes. + ingest.up = true; + clock.now += 1; + await transport.send(eventEnvelope("f")); + await vi.waitFor(() => expect(storedEventIds()).toEqual([])); + await vi.waitFor(() => + expect(ingest.sent).toEqual(["a", "d", "f", "a", "b", "c", "d", "e"]), + ); + + // Closed again: later envelopes go straight out. + await transport.send(eventEnvelope("g")); + expect(ingest.sent.at(-1)).toBe("g"); + expect(storedEventIds()).toEqual([]); + }); + + test("stores envelopes sent while the probe is in flight", async () => { + const clock = { now: 1_000_000 }; + const sent: unknown[] = []; + let answerProbe: () => void = noop; + let calls = 0; + const transport = makeStoringTransport( + { url: "http://127.0.0.1:9/api/0/envelope/", recordDroppedEvent: noop }, + () => ({ + send: (envelope) => { + sent.push(envelope[0].event_id); + calls += 1; + if (calls === 1) { + return Promise.reject(new Error("connect ECONNREFUSED")); + } + if (calls === 2) { + return new Promise((resolve) => { + answerProbe = () => resolve({ statusCode: 200 }); + }); + } + return Promise.resolve({ statusCode: 200 }); + }, + flush: () => Promise.resolve(true), + }), + () => clock.now, + ); + + await transport.send(eventEnvelope("a")); + clock.now += BREAKER_COOLDOWN_MS; + const probe = transport.send(eventEnvelope("probe")); + await transport.send(eventEnvelope("during")); + expect(sent).toEqual(["a", "probe"]); + expect(storedEventIds()).toEqual(["a", "during"]); + + answerProbe(); + await probe; + await vi.waitFor(() => expect(storedEventIds()).toEqual([])); + expect(sent).toEqual(["a", "probe", "a", "during"]); + }); + + test("drops runtime metrics instead of storing them", async () => { + const { clock, ingest, transport } = setup(); + const metrics = createEnvelope({ event_id: "m" }, [ + [{ type: "trace_metric" }, { items: [] }], + ] as never); + + // A failed send opens the breaker but keeps nothing. The send rejects, + // which the SDK client catches. + await expect(transport.send(metrics)).rejects.toThrow("ECONNREFUSED"); + expect(ingest.sent).toEqual(["m"]); + expect(storedEventIds()).toEqual([]); + + // With the breaker open, metrics skip the network and are dropped too, + // while other envelopes are still stored. + await expect(transport.send(metrics)).rejects.toThrow("shouldSend"); + await transport.send(eventEnvelope("a")); + expect(ingest.sent).toEqual(["m"]); + expect(storedEventIds()).toEqual(["a"]); + + // A metrics envelope can still be the probe that closes the breaker. + ingest.up = true; + clock.now += BREAKER_COOLDOWN_MS; + await transport.send(metrics); + await vi.waitFor(() => expect(storedEventIds()).toEqual([])); + expect(ingest.sent).toEqual(["m", "m", "a"]); + }); + + test("a response after a failure in the same batch doesn't drain more", async () => { + const store = createEnvelopeStore(); + for (let i = 0; i < 25; i++) { + await store.push(eventEnvelope(`s${i}`)); + } + let calls = 0; + const transport = makeStoringTransport( + { url: "http://127.0.0.1:9/api/0/envelope/", recordDroppedEvent: noop }, + () => ({ + // The first send of the batch fails at once; the rest answer a + // moment later, after the failure has opened the breaker. + send: async () => { + calls += 1; + if (calls === 1) { + throw new Error("connect ECONNREFUSED"); + } + await new Promise((resolve) => setTimeout(resolve, 5)); + return { statusCode: 200 }; + }, + flush: () => Promise.resolve(true), + }), + ); + + transport.sendStored(10); + + // One batch only: 9 delivered, the failed one requeued, 15 untouched. + await vi.waitFor(() => expect(calls).toBe(10)); + await new Promise((resolve) => setTimeout(resolve, 50)); + expect(calls).toBe(10); + expect(storedEventIds()).toHaveLength(16); + }); + + test("sendStored does nothing while the breaker is open", async () => { + const { ingest, transport } = setup(); + await transport.send(eventEnvelope("a")); + + ingest.up = true; + transport.sendStored(10); + + expect(ingest.sent).toEqual(["a"]); + expect(storedEventIds()).toEqual(["a"]); + }); +}); diff --git a/packages/cli/test/lib/telemetry/zstd-transport.e2e.test.ts b/packages/cli/test/lib/telemetry/zstd-transport.e2e.test.ts index 6ac061edf..70e1ed37e 100644 --- a/packages/cli/test/lib/telemetry/zstd-transport.e2e.test.ts +++ b/packages/cli/test/lib/telemetry/zstd-transport.e2e.test.ts @@ -7,13 +7,6 @@ * and the body decompresses back to a valid envelope. */ -import { createServer, type IncomingMessage } from "node:http"; -import { - type AddressInfo, - createServer as createTcpServer, - type Server as TcpServer, - type Socket, -} from "node:net"; import { promisify } from "node:util"; import { zstdDecompress } from "node:zlib"; import { createEnvelope } from "@sentry/core"; @@ -23,130 +16,16 @@ import { makeCompressedTransport, makeNodeTransportWithDeadline, } from "../../../src/lib/telemetry/zstd-transport.js"; - -/** No-op for SDK callbacks that require a function but return nothing meaningful. */ -function noop(): void { - // intentionally empty -} - -type CapturedRequest = { - headers: IncomingMessage["headers"]; - body: Buffer; -}; - -/** - * Track every server we start so they can be closed in teardown. - * Without this, a `let server` shared across tests is overwritten by - * the second test before the first one is closed — silent socket leak. - */ -// `http.Server` extends `net.Server`, so this holds both kinds. -const startedServers: TcpServer[] = []; - -/** - * Start a server that accepts connections and never answers, like an - * ingest endpoint behind a stalled network path. `closed` resolves when - * the client gives up and its socket closes. - */ -function startStalledIngest(): Promise<{ - url: string; - closed: Promise; -}> { - let onClosed: () => void = noop; - const closed = new Promise((resolve) => { - onClosed = resolve; - }); - const server = createTcpServer((socket: Socket) => { - socket.on("error", noop); - socket.on("close", () => onClosed()); - // Read (and drop) the request so the client's hang-up is seen. - socket.resume(); - }); - startedServers.push(server); - return new Promise((resolve) => { - server.listen(0, "127.0.0.1", () => { - const addr = server.address() as AddressInfo; - resolve({ - url: `http://127.0.0.1:${addr.port}/api/0/envelope/`, - closed, - }); - }); - }); -} - -/** - * Start an HTTP proxy that accepts the CONNECT tunnel and then goes - * silent. Routes the transport through the SDK's `makeNodeTransport`. - */ -function startStalledProxy(): Promise<{ url: string; closed: Promise }> { - let onClosed: () => void = noop; - const closed = new Promise((resolve) => { - onClosed = resolve; - }); - const server = createServer(); - server.on("connect", (_req, socket) => { - socket.on("error", noop); - socket.on("close", () => onClosed()); - // HTTP servers allow half-open sockets: close ours when the client - // hangs up, so `closed` proves the client released its socket. - socket.on("end", () => socket.end()); - socket.resume(); - socket.write("HTTP/1.1 200 Connection Established\r\n\r\n"); - }); - startedServers.push(server); - return new Promise((resolve) => { - server.listen(0, "127.0.0.1", () => { - const addr = server.address() as AddressInfo; - resolve({ url: `http://127.0.0.1:${addr.port}`, closed }); - }); - }); -} - -function startMockIngest( - responder: (req: IncomingMessage) => { - statusCode: number; - headers?: Record; - }, -): Promise<{ - url: string; - captures: CapturedRequest[]; -}> { - const captures: CapturedRequest[] = []; - const server = createServer((req, res) => { - const chunks: Buffer[] = []; - req.on("data", (c: Buffer) => chunks.push(c)); - req.on("end", () => { - captures.push({ headers: req.headers, body: Buffer.concat(chunks) }); - const response = responder(req); - res.writeHead(response.statusCode, response.headers ?? {}); - res.end(); - }); - }); - startedServers.push(server); - - return new Promise((resolve) => { - server.listen(0, "127.0.0.1", () => { - const addr = server.address() as AddressInfo; - resolve({ - url: `http://127.0.0.1:${addr.port}/api/0/envelope/`, - captures, - }); - }); - }); -} +import { + closeIngestServers, + noop, + startMockIngest, + startStalledIngest, + startStalledProxy, +} from "./ingest-servers.js"; describe("makeCompressedTransport (e2e)", () => { - afterEach(async () => { - // Close every server started in this test, in parallel. Lets the - // event loop drain naturally instead of relying on process exit. - await Promise.all( - startedServers.splice(0).map( - (s) => - new Promise((resolve) => { - s.close(() => resolve()); - }), - ), - ); - }); + afterEach(closeIngestServers); test("sends zstd-encoded envelope; server decompresses back to original", async () => { if (!hasZstdSupport()) { @@ -253,6 +132,8 @@ describe("makeCompressedTransport (e2e)", () => { }); describe("makeNodeTransportWithDeadline (e2e)", () => { + afterEach(closeIngestServers); + test("aborts a request the server never answers", async () => { const { url, closed } = await startStalledIngest(); const transport = makeNodeTransportWithDeadline(