diff --git a/packages/cli-repl/src/cli-repl-telemetry.spec.ts b/packages/cli-repl/src/cli-repl-telemetry.spec.ts index 89e289bf0b..2694664943 100644 --- a/packages/cli-repl/src/cli-repl-telemetry.spec.ts +++ b/packages/cli-repl/src/cli-repl-telemetry.spec.ts @@ -556,6 +556,26 @@ describe('CliRepl telemetry (integration)', function () { expect(payload.api_deprecation_errors).to.equal(true); }); }); + + it('deliver HEAD beacons without waiting for slow responses', async function () { + const testStartMs = Date.now(); + // The server delays responses by 5s. The fire-and-forget transport + // dispatches on write-finish and exits immediately; a transport that + // waited for responses would sit out the 2s flush timeout and bust + // the 1s post-start budget below. + setTelemetryDelay(5000); + await cliRepl.start(await testServer.connectionString(), {}); + this.timeout(Date.now() - testStartMs + 1000); // Exclude connection time from the 1s budget + input.write('use somedb;\n'); + input.write('exit\n'); + await waitBus(cliRepl.bus, 'mongosh:closed'); + // The fake server records requests on receipt, before delaying the + // response, so these are observable even though no response has + // been sent yet. + expect(requests).to.have.lengthOf.at.least(1); + expect(requests[0].req.method).to.equal('HEAD'); + expect(requests[0].req.headers['user-agent']).to.match(/^mongosh\//); + }); }); context('without network connectivity', function () { diff --git a/packages/cli-repl/src/cli-repl.ts b/packages/cli-repl/src/cli-repl.ts index e107472bd0..4f958a3636 100644 --- a/packages/cli-repl/src/cli-repl.ts +++ b/packages/cli-repl/src/cli-repl.ts @@ -52,6 +52,7 @@ import type { } from '@mongodb-js/devtools-proxy-support'; import { createFetch, + systemCA, useOrCreateAgent, } from '@mongodb-js/devtools-proxy-support'; import { fullDepthInspectOptions } from './format-output'; @@ -62,6 +63,16 @@ import { getDeviceIdForMongosh } from './device-id'; */ const CONNECTING = 'cli-repl.cli-repl.connecting'; +/** + * Sanitizes a list of User-Agent tag values (dropping nullish entries to + * '') and joins them the way the telemetry endpoint expects. + */ +function formatUserAgentTags(tags: (string | undefined)[]): string { + return tags + .map((s) => (s ?? '').replace(/[^a-zA-Z0-9./_-]/g, '_')) + .join('; '); +} + /** * The set of options taken by CliRepl instances. */ @@ -191,17 +202,15 @@ export class CliRepl implements MongoshIOProvider { return await this.getDeviceId(); })(), ]); - const userAgentTags = [ + const userAgentTags = formatUserAgentTags([ `mongosh/${version}`, os_type, os_release, os_arch, - os_linux_dist ?? '', - os_linux_release ?? '', + os_linux_dist, + os_linux_release, deviceId, - ] - .map((s) => s.replace(/[^a-zA-Z0-9./_-]/g, '_')) - .join('; '); + ]); return baseFetch(url, { ...init, headers: { @@ -689,13 +698,30 @@ export class CliRepl implements MongoshIOProvider { } async setupAnalytics(): Promise { + const { version }: { version: string } = require('../package.json'); + const { os_type, os_release, os_arch, os_linux_dist, os_linux_release } = + await this.getOsInfo(); const { analytics, telemetryEndpoint } = setupTelemetryAnalytics({ // `telemetryEndpoint` user config carries the production default. configuredTelemetryEndpoint: await this.getConfig('telemetryEndpoint'), - // includeDeviceId: false — device_id is already in the event payload, - // no need to duplicate it in the User-Agent header. - fetch: this.fetch({ includeDeviceId: false }), metadataPath: this.shellHomeDirectory.paths.shellLocalDataPath, + agent: this.agent, + // device_id is already in the event payload, so the User-Agent carries + // 'disabled' in the device slot instead of duplicating it. + userAgent: formatUserAgentTags([ + `mongosh/${version}`, + os_type, + os_release, + os_arch, + os_linux_dist, + os_linux_release, + 'disabled', + ]), + // The beacon builds its own agents, which would otherwise only trust + // Node's bundled roots; hand it the merged system CA list so custom-CA + // environments (corporate proxies, test sinks) keep working. systemCA() + // is pre-warmed at startup, so this is a cache hit. + tlsCa: (await systemCA()).ca, }); this.toggleableAnalytics = analytics; // Record the resolved endpoint so logging can decide whether to log full diff --git a/packages/cli-repl/src/setup-analytics.spec.ts b/packages/cli-repl/src/setup-analytics.spec.ts index 6bdab15ecc..fc1a6a1c87 100644 --- a/packages/cli-repl/src/setup-analytics.spec.ts +++ b/packages/cli-repl/src/setup-analytics.spec.ts @@ -6,7 +6,12 @@ import { ToggleableAnalytics, } from '@mongosh/logging'; import type { TelemetryEvent } from '@mongosh/logging'; -import { setupTelemetryAnalytics } from './setup-analytics'; +import type { AgentWithInitialize } from '@mongodb-js/devtools-proxy-support'; +import { useOrCreateAgent } from '@mongodb-js/devtools-proxy-support'; +import { + resolveTelemetryAgent, + setupTelemetryAnalytics, +} from './setup-analytics'; const identifyEvent: TelemetryEvent = { name: 'Identify', @@ -30,86 +35,133 @@ const identifyEvent: TelemetryEvent = { }, }; -describe('setupTelemetryAnalytics', function () { - const metadataPath = os.tmpdir(); - // A fetch stub; these tests never actually track()/send, they only inspect - // how the analytics sink is constructed. - const fetch = () => Promise.resolve(new Response()); +describe('setup-analytics', function () { + describe('setupTelemetryAnalytics', function () { + const metadataPath = os.tmpdir(); - let savedEnvEndpoint: string | undefined; - beforeEach(function () { - savedEnvEndpoint = process.env.MONGOSH_TELEMETRY_ENDPOINT; - delete process.env.MONGOSH_TELEMETRY_ENDPOINT; - }); - afterEach(function () { - if (savedEnvEndpoint === undefined) { + let savedEnvEndpoint: string | undefined; + beforeEach(function () { + savedEnvEndpoint = process.env.MONGOSH_TELEMETRY_ENDPOINT; delete process.env.MONGOSH_TELEMETRY_ENDPOINT; - } else { - process.env.MONGOSH_TELEMETRY_ENDPOINT = savedEnvEndpoint; + }); + afterEach(function () { + if (savedEnvEndpoint === undefined) { + delete process.env.MONGOSH_TELEMETRY_ENDPOINT; + } else { + process.env.MONGOSH_TELEMETRY_ENDPOINT = savedEnvEndpoint; + } + }); + + function setup( + params: Partial[0]> = {} + ) { + return setupTelemetryAnalytics({ + configuredTelemetryEndpoint: '', + metadataPath, + ...params, + }); } - }); - function setup( - params: Partial[0]> = {} - ) { - return setupTelemetryAnalytics({ - configuredTelemetryEndpoint: '', - fetch: fetch as any, - metadataPath, - ...params, + it('returns a no-op sink when no endpoint is configured', function () { + const { analytics, telemetryEndpoint } = setup(); + expect(telemetryEndpoint).to.equal(''); + expect(analytics).to.be.instanceOf(ToggleableAnalytics); + // No endpoint -> nothing to send to. Telemetry is not disabled here; + // events are still logged locally, they just have no destination. + expect(analytics._target).to.be.instanceOf(NoopAnalytics); }); - } - it('returns a no-op sink when no endpoint is configured', function () { - const { analytics, telemetryEndpoint } = setup(); - expect(telemetryEndpoint).to.equal(''); - expect(analytics).to.be.instanceOf(ToggleableAnalytics); - // No endpoint -> nothing to send to. Telemetry is not disabled here; - // events are still logged locally, they just have no destination. - expect(analytics._target).to.be.instanceOf(NoopAnalytics); - }); + it('creates a telemetry client when an endpoint is configured via user config', function () { + const { analytics, telemetryEndpoint } = setup({ + configuredTelemetryEndpoint: 'https://config.example/events', + }); + expect(telemetryEndpoint).to.equal('https://config.example/events'); + expect(analytics._target).to.be.instanceOf(ThrottledAnalytics); + }); - it('creates a telemetry client when an endpoint is configured via user config', function () { - const { analytics, telemetryEndpoint } = setup({ - configuredTelemetryEndpoint: 'https://config.example/events', + it('uses MONGOSH_TELEMETRY_ENDPOINT over the configured default', function () { + process.env.MONGOSH_TELEMETRY_ENDPOINT = 'https://env.example/events'; + const { telemetryEndpoint, analytics } = setup({ + configuredTelemetryEndpoint: 'https://config.example/events', + }); + expect(telemetryEndpoint).to.equal('https://env.example/events'); + expect(analytics._target).to.be.instanceOf(ThrottledAnalytics); }); - expect(telemetryEndpoint).to.equal('https://config.example/events'); - expect(analytics._target).to.be.instanceOf(ThrottledAnalytics); - }); - it('uses MONGOSH_TELEMETRY_ENDPOINT over the configured default', function () { - process.env.MONGOSH_TELEMETRY_ENDPOINT = 'https://env.example/events'; - const { telemetryEndpoint, analytics } = setup({ - configuredTelemetryEndpoint: 'https://config.example/events', + it('is disabled when every source resolves to an empty endpoint', function () { + process.env.MONGOSH_TELEMETRY_ENDPOINT = ''; + const { telemetryEndpoint, analytics } = setup({ + configuredTelemetryEndpoint: '', + }); + expect(telemetryEndpoint).to.equal(''); + expect(analytics._target).to.be.instanceOf(NoopAnalytics); }); - expect(telemetryEndpoint).to.equal('https://env.example/events'); - expect(analytics._target).to.be.instanceOf(ThrottledAnalytics); - }); - it('is disabled when every source resolves to an empty endpoint', function () { - process.env.MONGOSH_TELEMETRY_ENDPOINT = ''; - const { telemetryEndpoint, analytics } = setup({ - configuredTelemetryEndpoint: '', + it('never constructs a network-capable sink when no endpoint is configured', async function () { + const { analytics } = setup({ + configuredTelemetryEndpoint: '', + }); + // With no endpoint the target is a NoopAnalytics — no beacon exists, + // so tracking and flushing can never produce a network request. + expect(analytics._target).to.be.instanceOf(NoopAnalytics); + analytics.enable(); + analytics.track(identifyEvent); + await analytics.flush(); // must not throw + expect(analytics._target).to.be.instanceOf(NoopAnalytics); }); - expect(telemetryEndpoint).to.equal(''); - expect(analytics._target).to.be.instanceOf(NoopAnalytics); }); - it('never calls fetch when no endpoint is configured', async function () { - let fetchCount = 0; - const { analytics } = setup({ - configuredTelemetryEndpoint: '', - fetch: (() => { - fetchCount++; - return Promise.resolve(new Response()); - }) as any, + describe('resolveTelemetryAgent', function () { + const createdAgents: (AgentWithInitialize | undefined)[] = []; + + afterEach(function () { + // Mirrors cli-repl.ts (which destroys its shared agent without + // awaiting); none of these agents ever open a real connection. + for (const agent of createdAgents.splice(0)) { + agent?.destroy(); + } + }); + + it('return undefined when the agent has no proxy configured for the endpoint', function () { + const agent = useOrCreateAgent({}); + createdAgents.push(agent); + const resolved = resolveTelemetryAgent( + agent, + 'https://telemetry.example.com' + ); + expect(resolved).to.equal(undefined); + }); + + it('return the agent unchanged when a proxy is configured for the endpoint', function () { + const agent = useOrCreateAgent({ + proxy: 'http://proxy.example.com:8080', + }); + createdAgents.push(agent); + const resolved = resolveTelemetryAgent( + agent, + 'https://telemetry.example.com' + ); + expect(resolved).to.equal(agent); + }); + + it('return undefined when there is no agent to resolve', function () { + const resolved = resolveTelemetryAgent( + undefined, + 'https://telemetry.example.com' + ); + expect(resolved).to.equal(undefined); + }); + + it('return undefined for an unparsable telemetry endpoint', function () { + // Use an agent with proxy configured to trigger the code path that + // parses the target URL in proxyForUrl(). With a malformed endpoint, + // this should throw; the fix wraps it in try/catch and returns undefined. + const agent = useOrCreateAgent({ + proxy: 'http://proxy.example.com:8080', + }); + createdAgents.push(agent); + const resolved = resolveTelemetryAgent(agent, 'not a url'); + expect(resolved).to.equal(undefined); }); - // Enable the queue so tracked events are forwarded to the target, then - // flush — with no endpoint the target is a NoopAnalytics, so no request - // is ever made. - analytics.enable(); - analytics.track(identifyEvent); - await analytics.flush(); - expect(fetchCount).to.equal(0); }); }); diff --git a/packages/cli-repl/src/setup-analytics.ts b/packages/cli-repl/src/setup-analytics.ts index 10278a003a..2727e944e4 100644 --- a/packages/cli-repl/src/setup-analytics.ts +++ b/packages/cli-repl/src/setup-analytics.ts @@ -1,9 +1,12 @@ -import type { RequestInit, Response } from '@mongodb-js/devtools-proxy-support'; +import type { AgentWithInitialize } from '@mongodb-js/devtools-proxy-support'; +import { useOrCreateAgent } from '@mongodb-js/devtools-proxy-support'; import { ThrottledAnalytics, ToggleableAnalytics, TelemetryClient, + FireAndForgetBeacon, } from '@mongosh/logging'; +import path from 'path'; /** * ThrottledAnalytics caps events to protect against high-frequency @@ -17,10 +20,32 @@ export type SetupTelemetryAnalyticsParams = { * default. Used as the lowest-priority source when resolving the endpoint. */ configuredTelemetryEndpoint: string; - /** Proxy-aware fetch used to deliver telemetry events. */ - fetch: (url: string, init?: RequestInit) => Promise; /** Directory used to persist cross-session throttle state. */ metadataPath: string; + /** + * Proxy-aware agent shared with the rest of mongosh. This is always + * present in a real mongosh — `useOrCreateAgent` is called without a + * target in cli-repl.ts, and without one it always builds an agent rather + * than returning undefined. The telemetry beacon only reuses it + * (via {@link resolveTelemetryAgent}) when a proxy actually applies to the + * telemetry endpoint; otherwise it builds its own resuming keep-alive + * agent so TLS session resumption and the local DNS cache stay in effect. + * Note that when a proxy agent *is* used, the beacon's `dispatched` + * guarantee ("bytes reached the kernel") is a little weaker: the + * client-visible connect event can fire once the tunnel to the proxy + * is established, ahead of the end-to-end TLS handshake completing — + * acceptable for best-effort telemetry. + */ + agent?: AgentWithInitialize; + /** User-Agent header value attached to every telemetry request. */ + userAgent?: string; + /** + * Merged CA list (system store + bundled roots, from devtools-proxy-support's + * systemCA). The beacon builds its own agents, which would otherwise only + * trust Node's bundled roots — this keeps custom-CA environments + * (corporate proxies, test sinks) working. + */ + tlsCa?: string; }; export type SetupTelemetryAnalyticsResult = { @@ -35,6 +60,41 @@ export type SetupTelemetryAnalyticsResult = { telemetryEndpoint: string; }; +/** + * Resolves the shared proxy-aware agent against the telemetry endpoint for + * use by the fire-and-forget transport. + * + * `useOrCreateAgent(agent, telemetryEndpoint, true)`, given an *existing* + * agent instance plus a target and `useTargetRegardlessOfExistingAgent: + * true`, re-checks that agent's own `proxyOptions` against the target and: + * - returns `undefined` when they resolve to no proxy for that URL — the + * fire-and-forget transport then builds its own resuming keep-alive + * agent (TLS session resumption + DNS cache) instead of reusing the + * general-purpose one, both of which would otherwise be dead code; + * - returns the agent unchanged when a proxy *does* apply, so proxy + * environments keep working. + * (Verified against node_modules/@mongodb-js/devtools-proxy-support + * dist/agent.js: `useOrCreateAgent` branches on `isExistingAgentInstance` + * (`'createConnection' in options`, true for any agent produced by + * `createAgent`/`useOrCreateAgent`) and, on that branch, returns `undefined` + * exactly when `useTargetRegardlessOfExistingAgent && target !== undefined + * && agent.proxyOptions && !proxyForUrl(agent.proxyOptions, target)`.) + */ +export function resolveTelemetryAgent( + agent: AgentWithInitialize | undefined, + telemetryEndpoint: string +): AgentWithInitialize | undefined { + if (!agent) return undefined; + try { + return useOrCreateAgent(agent, telemetryEndpoint, true); + } catch { + // A malformed endpoint must not break analytics setup; the fire-and-forget + // beacon then builds its own agent and telemetry sends fail silently + // downstream, consistent with the never-throw contract. + return undefined; + } +} + /** * Build the analytics sink for a mongosh session. * @@ -46,8 +106,10 @@ export type SetupTelemetryAnalyticsResult = { */ export function setupTelemetryAnalytics({ configuredTelemetryEndpoint, - fetch, metadataPath, + agent, + userAgent, + tlsCa, }: SetupTelemetryAnalyticsParams): SetupTelemetryAnalyticsResult { // Resolve the telemetry endpoint: MONGOSH_TELEMETRY_ENDPOINT environment // variable > `telemetryEndpoint` user config (which carries the prod default). @@ -56,16 +118,27 @@ export function setupTelemetryAnalytics({ if (!telemetryEndpoint) { return { analytics: new ToggleableAnalytics(), telemetryEndpoint: '' }; } + // Fire-and-forget transport (MONGOSH-3454): resolves sends once the + // request is written to an established socket instead of waiting for the + // response, so telemetry can never delay mongosh exit. Owns its own + // health policy (adaptive timeout, circuit breaker) and persists TLS + // session tickets next to the throttle state for cross-session resumption. + const beacon = new FireAndForgetBeacon({ + agent: resolveTelemetryAgent(agent, telemetryEndpoint), + defaultHeaders: userAgent ? { 'User-Agent': userAgent } : {}, + tlsOptions: tlsCa ? { ca: tlsCa } : undefined, + sessionStorePath: path.join(metadataPath, 'telemetry-tls-sessions.json'), + }); return { telemetryEndpoint, // ThrottledAnalytics wraps TelemetryClient target and gates every // track() call before it reaches it. The timeframe defaults to 60s. // Once the cap is hit, further events within the same window // are silently dropped and TelemetryClient.track() - // (and its underlying fetch) is never called. + // (and the beacon behind it) is never called. analytics: new ToggleableAnalytics( new ThrottledAnalytics({ - target: new TelemetryClient(telemetryEndpoint, fetch), + target: new TelemetryClient(telemetryEndpoint, beacon), throttle: { rate: TELEMETRY_THROTTLE_RATE, metadataPath, diff --git a/packages/logging/src/beacon.ts b/packages/logging/src/beacon.ts new file mode 100644 index 0000000000..4b8fbb6082 --- /dev/null +++ b/packages/logging/src/beacon.ts @@ -0,0 +1,40 @@ +// Generous enough number for a fire-and-forget event. Adjust freely if needed. +export const REQUEST_TIMEOUT_MS = 5_000; + +/** + * The result of a beacon send. `send()` never rejects; failures are reported + * as `error` outcomes so callers do not need try/catch on the send path. + * + * - 'dispatched': the request was fully written to an established connection; + * no response was awaited (fire-and-forget implementations). + * - 'response': a response was received (implementations that wait for one). + * - 'error': the request could not be delivered. + * - 'suppressed': nothing was sent — the implementation's circuit breaker is + * open because the endpoint has been consistently failing. + */ +export type BeaconOutcome = + | { kind: 'dispatched'; durationMs: number } + | { kind: 'response'; status: number; durationMs: number } + | { kind: 'error'; error: Error; durationMs: number } + | { kind: 'suppressed' }; + +/** + * Transport used by TelemetryClient to deliver HEAD beacons to the telemetry + * endpoint. The client owns the event format and serialization; implementations + * own every communication concern: sockets, pooling, request timeouts, + * health tracking, and giving up on a dead endpoint. + */ +export interface Beacon { + send(url: string, headers: Record): Promise; + /** + * Impending-shutdown hook: perform any last bounded I/O (e.g. persisting + * TLS session tickets). Callers race it against a short deadline and it is + * invoked after in-flight sends settle; it must resolve quickly and never + * reject. + */ + flush?(): Promise; + /** Optionally pre-establish a connection (DNS/TCP/TLS) before the first send. */ + warmUp?(url: string): void; + /** Optionally release held resources (sockets, agents). */ + close?(): void; +} diff --git a/packages/logging/src/fire-and-forget-beacon.spec.ts b/packages/logging/src/fire-and-forget-beacon.spec.ts new file mode 100644 index 0000000000..b0e53e1df6 --- /dev/null +++ b/packages/logging/src/fire-and-forget-beacon.spec.ts @@ -0,0 +1,561 @@ +import { expect } from 'chai'; +import http from 'http'; +import { once } from 'events'; +import { spawn } from 'child_process'; +import path from 'path'; +import type { AddressInfo } from 'net'; +import fs from 'fs'; +import os from 'os'; +import type tls from 'tls'; +import { createServer as createHttpsServer } from 'https'; +import type { Server as HttpsServer } from 'https'; +import { + FireAndForgetBeacon, + createCachedLookup, +} from './fire-and-forget-beacon'; + +describe('FireAndForgetBeacon', function () { + let beacon: FireAndForgetBeacon; + + afterEach(function () { + beacon?.close(); + }); + + context('against a responding server', function () { + let srv: http.Server; + let baseUrl: string; + let requests: http.IncomingMessage[]; + let connections: number; + + beforeEach(async function () { + requests = []; + connections = 0; + srv = http + .createServer((req, res) => { + requests.push(req); + // Content-Length must be explicit: Node's client-side HTTP parser + // decides keep-alive eligibility from response framing headers + // before it learns (from the request method) that a HEAD response + // has no body, so a HEAD response with neither Content-Length nor + // Transfer-Encoding is always treated as non-keep-alive and the + // socket is destroyed instead of pooled. + res.writeHead(200, { 'Content-Length': '0' }); + res.end(); + }) + .on('connection', () => { + connections++; + }) + .listen(0); + await once(srv, 'listening'); + baseUrl = `http://localhost:${(srv.address() as AddressInfo).port}`; + }); + + afterEach(async function () { + srv.close(); + await once(srv, 'close'); + }); + + it('send a HEAD request with the provided headers', async function () { + beacon = new FireAndForgetBeacon(); + // Subscribe before sending: `dispatched` may resolve before or after + // the server has parsed the request, so a post-send once() could hang. + const requestReceived = once(srv, 'request'); + const outcome = await beacon.send(`${baseUrl}/v1/test`, { + Cookie: 'mge=abc', + }); + expect(outcome.kind).to.equal('dispatched'); + expect(outcome) + .to.have.property('durationMs') + .that.is.a('number') + .and.is.at.least(0); + await requestReceived; + expect(requests).to.have.lengthOf(1); + expect(requests[0].method).to.equal('HEAD'); + expect(requests[0].url).to.equal('/v1/test'); + expect(requests[0].headers.cookie).to.equal('mge=abc'); + }); + + it('merge default headers into every request', async function () { + beacon = new FireAndForgetBeacon({ + defaultHeaders: { 'User-Agent': 'mongosh/9.9.9' }, + }); + const requestReceived = once(srv, 'request'); + await beacon.send(`${baseUrl}/v1/test`, { Cookie: 'mge=abc' }); + await requestReceived; + expect(requests).to.have.lengthOf(1); + expect(requests[0].headers['user-agent']).to.equal('mongosh/9.9.9'); + expect(requests[0].headers.cookie).to.equal('mge=abc'); + }); + + it('reuse the keep-alive connection across sequential sends', async function () { + beacon = new FireAndForgetBeacon(); + await beacon.send(`${baseUrl}/v1/one`, {}); + // Wait for the first response to complete so the socket returns to the pool. + await new Promise((resolve) => setTimeout(resolve, 100)); + await beacon.send(`${baseUrl}/v1/two`, {}); + await new Promise((resolve) => setTimeout(resolve, 100)); + expect(requests).to.have.lengthOf(2); + expect(connections).to.equal(1); + }); + }); + + context('against a server that never responds', function () { + let srv: http.Server; + let baseUrl: string; + let seenRequests: number; + + beforeEach(async function () { + seenRequests = 0; + // Accepts connections and requests but never sends a response. + srv = http + .createServer(() => { + seenRequests++; + }) + .listen(0); + await once(srv, 'listening'); + baseUrl = `http://localhost:${(srv.address() as AddressInfo).port}`; + }); + + afterEach(async function () { + srv.closeAllConnections(); + srv.close(); + await once(srv, 'close'); + }); + + it('resolve as dispatched without waiting for the server to respond', async function () { + beacon = new FireAndForgetBeacon(); + const start = Date.now(); + const outcome = await beacon.send(`${baseUrl}/v1/test`, { + Cookie: 'mge=abc', + }); + expect(outcome.kind).to.equal('dispatched'); + expect(Date.now() - start).to.be.lessThan(1_000); + }); + + it('open parallel connections for concurrent sends instead of queueing behind a stalled response', async function () { + beacon = new FireAndForgetBeacon(); + const start = Date.now(); + // Subscribe before sending: `dispatched` resolves from the client-side + // 'finish'/'connect' events, whose microtask continuations run before + // the event loop gets to the server's own (separate) socket callbacks, + // so asserting on `seenRequests` immediately after would be racy. + let received = 0; + const bothRequestsReceived = new Promise((resolve) => { + srv.on('request', () => { + received++; + if (received === 2) resolve(); + }); + }); + const outcomes = await Promise.all([ + beacon.send(`${baseUrl}/v1/one`, {}), + beacon.send(`${baseUrl}/v1/two`, {}), + ]); + expect(outcomes.map(({ kind }) => kind)).to.deep.equal([ + 'dispatched', + 'dispatched', + ]); + expect(Date.now() - start).to.be.lessThan(1_000); + await bothRequestsReceived; + expect(seenRequests).to.equal(2); + }); + + it('let the process exit while the server is still holding the request open', async function () { + // Spawns a child that sends a beacon to this blackhole server and then + // reaches the end of its script. Because sockets are unref'd, the child + // must exit on its own; a ref'd socket would hang it until the timeout. + const fixture = path.resolve( + __dirname, + '..', + 'test', + 'fixtures', + 'beacon-exit-fixture.ts' + ); + const child = spawn( + process.execPath, + [ + '--require', + 'ts-node/register/transpile-only', + fixture, + String((srv.address() as AddressInfo).port), + ], + { + cwd: path.resolve(__dirname, '..'), + stdio: ['ignore', 'pipe', 'pipe'], + } + ); + let stdout = ''; + let stderr = ''; + child.stdout.setEncoding('utf8').on('data', (chunk) => { + stdout += chunk; + }); + child.stderr.setEncoding('utf8').on('data', (chunk) => { + stderr += chunk; + }); + const [code] = await once(child, 'exit'); + expect(stderr).to.equal(''); + expect(stdout).to.include('"kind":"dispatched"'); + expect(code).to.equal(0); + }); + }); + + context('against an unreachable endpoint', function () { + it('resolve with an error outcome when the connection is refused', async function () { + // Grab a port that is momentarily free, then close the server so + // nothing is listening on it. + const srv = http.createServer().listen(0); + await once(srv, 'listening'); + const port = (srv.address() as AddressInfo).port; + srv.close(); + await once(srv, 'close'); + + beacon = new FireAndForgetBeacon(); + const outcome = await beacon.send(`http://localhost:${port}/v1/test`, {}); + expect(outcome.kind).to.equal('error'); + expect(outcome).to.have.nested.property('error.code', 'ECONNREFUSED'); + }); + + it('resolve with an error outcome for an unparsable URL', async function () { + beacon = new FireAndForgetBeacon(); + const outcome = await beacon.send('not a url', {}); + expect(outcome.kind).to.equal('error'); + expect(outcome).to.have.nested.property('error.name', 'TypeError'); + }); + + it('resolve with an error outcome for an unsupported URL scheme', async function () { + beacon = new FireAndForgetBeacon(); + const outcome = await beacon.send('ftp://example.com/v1/test', {}); + expect(outcome.kind).to.equal('error'); + expect(outcome).to.have.nested.property( + 'error.code', + 'ERR_INVALID_PROTOCOL' + ); + }); + }); + + context('health tracking', function () { + let liveSrv: http.Server; + let liveUrl: string; + let refusedUrl: string; + + beforeEach(async function () { + liveSrv = http + .createServer((req, res) => { + res.writeHead(200); + res.end(); + }) + .listen(0); + await once(liveSrv, 'listening'); + liveUrl = `http://localhost:${ + (liveSrv.address() as AddressInfo).port + }/v1/live`; + + // Grab a port that is momentarily free, then free it again. + const gone = http.createServer().listen(0); + await once(gone, 'listening'); + const refusedPort = (gone.address() as AddressInfo).port; + gone.close(); + await once(gone, 'close'); + refusedUrl = `http://localhost:${refusedPort}/v1/refused`; + }); + + afterEach(async function () { + liveSrv.close(); + await once(liveSrv, 'close'); + }); + + it('use the default timeout until enough samples are collected', function () { + beacon = new FireAndForgetBeacon(); + expect(beacon.currentTimeoutMs()).to.equal(5_000); + }); + + it('tighten the timeout from observed dispatch durations', async function () { + beacon = new FireAndForgetBeacon({ + timeouts: { minSamples: 3, minMs: 10 }, + }); + const outcomes = [ + await beacon.send(liveUrl, {}), + await beacon.send(liveUrl, {}), + await beacon.send(liveUrl, {}), + ]; + expect(outcomes.map(({ kind }) => kind)).to.deep.equal([ + 'dispatched', + 'dispatched', + 'dispatched', + ]); + expect(beacon.currentTimeoutMs()).to.be.lessThan(5_000); + expect(beacon.currentTimeoutMs()).to.be.at.least(10); + }); + + it('open the breaker after the configured number of consecutive failures', async function () { + beacon = new FireAndForgetBeacon({ breaker: { threshold: 2 } }); + const first = await beacon.send(refusedUrl, {}); + const second = await beacon.send(refusedUrl, {}); + const third = await beacon.send(refusedUrl, {}); + expect(first.kind).to.equal('error'); + expect(second.kind).to.equal('error'); + expect(third.kind).to.equal('suppressed'); + }); + + it('allow a probe after the cooldown and reopen on its failure', async function () { + beacon = new FireAndForgetBeacon({ + breaker: { threshold: 1, cooldownMs: 50 }, + }); + const initial = await beacon.send(refusedUrl, {}); + const whileOpen = await beacon.send(refusedUrl, {}); + await new Promise((resolve) => setTimeout(resolve, 75)); + const probe = await beacon.send(refusedUrl, {}); + const reopened = await beacon.send(refusedUrl, {}); + expect(initial.kind).to.equal('error'); + expect(whileOpen.kind).to.equal('suppressed'); + expect(probe.kind).to.equal('error'); // the probe actually went out + expect(reopened.kind).to.equal('suppressed'); // and its failure reopened the breaker + }); + + it('reset the failure count after a successful dispatch', async function () { + beacon = new FireAndForgetBeacon({ breaker: { threshold: 2 } }); + const fail1 = await beacon.send(refusedUrl, {}); + const ok = await beacon.send(liveUrl, {}); + const fail2 = await beacon.send(refusedUrl, {}); + // Without the reset, fail2 would have been the second consecutive + // failure and this send would be suppressed; with it, the breaker + // only opens as fail3 completes. + const fail3 = await beacon.send(refusedUrl, {}); + const suppressed = await beacon.send(refusedUrl, {}); + expect(fail1.kind).to.equal('error'); + expect(ok.kind).to.equal('dispatched'); + expect(fail2.kind).to.equal('error'); + expect(fail3.kind).to.equal('error'); + expect(suppressed.kind).to.equal('suppressed'); + }); + + it('let only one probe through when concurrent sends race the half-open window', async function () { + beacon = new FireAndForgetBeacon({ + breaker: { threshold: 1, cooldownMs: 50 }, + }); + const initial = await beacon.send(refusedUrl, {}); + expect(initial.kind).to.equal('error'); // opens the breaker + await new Promise((resolve) => setTimeout(resolve, 75)); // cooldown expires + // All three send() calls execute synchronously before the probe's + // connection attempt can resolve, so exactly one may become the probe. + const outcomes = await Promise.all([ + beacon.send(refusedUrl, {}), + beacon.send(refusedUrl, {}), + beacon.send(refusedUrl, {}), + ]); + const kinds = outcomes.map(({ kind }) => kind).sort(); + expect(kinds).to.deep.equal(['error', 'suppressed', 'suppressed']); + }); + }); + + context('createCachedLookup', function () { + it('resolve from the cache within the TTL', async function () { + let baseCalls = 0; + const base = ((hostname: any, options: any, callback: any) => { + baseCalls++; + callback(null, '127.0.0.1', 4); + }) as any; + // eslint-disable-next-line @typescript-eslint/no-unsafe-argument + const lookup = createCachedLookup(3_600_000, base); + + const first = await new Promise((resolve) => + lookup('example.com', {}, (...args: unknown[]) => resolve(args)) + ); + const second = await new Promise((resolve) => + lookup('example.com', {}, (...args: unknown[]) => resolve(args)) + ); + + expect(first).to.deep.equal([null, '127.0.0.1', 4]); + expect(second).to.deep.equal([null, '127.0.0.1', 4]); + expect(baseCalls).to.equal(1); + }); + + it('fall back to the base lookup after the TTL expires', async function () { + let baseCalls = 0; + const base = ((hostname: any, options: any, callback: any) => { + baseCalls++; + callback(null, '127.0.0.1', 4); + }) as any; + // eslint-disable-next-line @typescript-eslint/no-unsafe-argument + const lookup = createCachedLookup(-1, base); // everything is expired + + await new Promise((resolve) => lookup('example.com', {}, resolve)); + await new Promise((resolve) => lookup('example.com', {}, resolve)); + + expect(baseCalls).to.equal(2); + }); + + it('bypass the cache for all-addresses lookups', async function () { + let baseCalls = 0; + const base = ((hostname: any, options: any, callback: any) => { + baseCalls++; + callback(null, [{ address: '127.0.0.1', family: 4 }]); + }) as any; + // eslint-disable-next-line @typescript-eslint/no-unsafe-argument + const lookup = createCachedLookup(3_600_000, base); + + await new Promise((resolve) => + lookup('example.com', { all: true }, resolve) + ); + await new Promise((resolve) => + lookup('example.com', { all: true }, resolve) + ); + + expect(baseCalls).to.equal(2); + }); + }); + + context('warm-up', function () { + let srv: http.Server; + let baseUrl: string; + let paths: string[]; + let connections: number; + + beforeEach(async function () { + paths = []; + connections = 0; + srv = http + .createServer((req, res) => { + paths.push(req.url ?? ''); + // Content-Length must be explicit for the same reason as above: + // otherwise the HEAD response is treated as non-keep-alive and the + // warm-up socket never makes it back to the pool for reuse. + res.writeHead(200, { 'Content-Length': '0' }); + res.end(); + }) + .on('connection', () => { + connections++; + }) + .listen(0); + await once(srv, 'listening'); + baseUrl = `http://localhost:${(srv.address() as AddressInfo).port}`; + }); + + afterEach(async function () { + srv.close(); + await once(srv, 'close'); + }); + + it('establish the connection during warm-up so the first send reuses it', async function () { + beacon = new FireAndForgetBeacon(); + beacon.warmUp(`${baseUrl}/warm-up`); + // Wait for the warm-up response to complete and return to the pool. + await new Promise((resolve) => setTimeout(resolve, 100)); + await beacon.send(`${baseUrl}/v1/test`, {}); + await new Promise((resolve) => setTimeout(resolve, 100)); + expect(paths).to.deep.equal(['/warm-up', '/v1/test']); + expect(connections).to.equal(1); + }); + }); + + context('over TLS', function () { + let srv: HttpsServer; + let baseUrl: string; + let reusedFlags: boolean[]; + let dir: string; + let storePath: string; + + beforeEach(async function () { + reusedFlags = []; + dir = fs.mkdtempSync(path.join(os.tmpdir(), 'beacon-tls-')); + storePath = path.join(dir, 'sessions.json'); + // Any syntactically valid key/cert pair works here: clients use + // rejectUnauthorized: false because these tests target the handshake + // lifecycle, not certificate validation. + const certDir = path.resolve( + __dirname, + '..', + '..', + 'testing', + 'certificates', + 'partial-trust-chain' + ); + srv = createHttpsServer( + { + key: fs.readFileSync(path.join(certDir, 'key.pem')), + cert: fs.readFileSync(path.join(certDir, 'cert.pem')), + }, + (req, res) => { + reusedFlags.push((req.socket as tls.TLSSocket).isSessionReused()); + res.writeHead(200); + res.end(); + } + ).listen(0); + await once(srv, 'listening'); + baseUrl = `https://localhost:${(srv.address() as AddressInfo).port}`; + }); + + afterEach(async function () { + srv.close(); + await once(srv, 'close'); + fs.rmSync(dir, { recursive: true, force: true }); + }); + + it('resolve as dispatched only after the TLS handshake completes', async function () { + beacon = new FireAndForgetBeacon({ + tlsOptions: { rejectUnauthorized: false }, + }); + const requestReceived = once(srv, 'request'); + const outcome = await beacon.send(`${baseUrl}/v1/test`, { + Cookie: 'mge=tls', + }); + expect(outcome.kind).to.equal('dispatched'); + await requestReceived; + expect(reusedFlags).to.deep.equal([false]); + }); + + it('resume the TLS session from the persisted ticket in a fresh beacon', async function () { + const first = new FireAndForgetBeacon({ + tlsOptions: { rejectUnauthorized: false }, + sessionStorePath: storePath, + }); + const firstReceived = once(srv, 'request'); + const firstOutcome = await first.send(`${baseUrl}/v1/one`, {}); + expect(firstOutcome.kind).to.equal('dispatched'); + await firstReceived; + // The shutdown hook: waits for the in-flight TLS 1.3 ticket (which + // arrives after `dispatched`) and for the store write to land — this is + // exactly what TelemetryClient.flush() does for a real mongosh exit. + await first.flush(); + expect(fs.existsSync(storePath)).to.equal(true); + first.close(); + + // Fresh instance = simulates the next mongosh process. + beacon = new FireAndForgetBeacon({ + tlsOptions: { rejectUnauthorized: false }, + sessionStorePath: storePath, + }); + const secondReceived = once(srv, 'request'); + const secondOutcome = await beacon.send(`${baseUrl}/v1/two`, {}); + expect(secondOutcome.kind).to.equal('dispatched'); + await secondReceived; + + expect(reusedFlags).to.deep.equal([false, true]); + }); + + it('resolve flush() quickly when the handshake never completes', async function () { + // Grab a port that is momentarily free, then close the server so + // nothing is listening on it: the connection is refused before any + // TLS handshake starts, so no session ticket can ever arrive and + // flush() must not pay the ticket grace period. + const gone = http.createServer().listen(0); + await once(gone, 'listening'); + const port = (gone.address() as AddressInfo).port; + gone.close(); + await once(gone, 'close'); + + beacon = new FireAndForgetBeacon({ + tlsOptions: { rejectUnauthorized: false }, + sessionStorePath: storePath, + }); + const outcome = await beacon.send( + `https://localhost:${port}/v1/test`, + {} + ); + expect(outcome.kind).to.equal('error'); + + const start = Date.now(); + await beacon.flush(); + expect(Date.now() - start).to.be.lessThan(90); + }); + }); +}); diff --git a/packages/logging/src/fire-and-forget-beacon.ts b/packages/logging/src/fire-and-forget-beacon.ts new file mode 100644 index 0000000000..69b2f49271 --- /dev/null +++ b/packages/logging/src/fire-and-forget-beacon.ts @@ -0,0 +1,514 @@ +import type dns from 'dns'; +import type http from 'http'; +import type https from 'https'; +import type tls from 'tls'; +import type net from 'net'; +import type { Duplex } from 'stream'; +import type { Beacon, BeaconOutcome } from './beacon'; +import { REQUEST_TIMEOUT_MS } from './beacon'; + +/** Ref'd post-dispatch drain window in flush(); see flush() for rationale. */ +const FLUSH_SETTLE_DELAY_MS = 10; +/** Bounded wait for an in-flight TLS 1.3 session ticket during flush(). */ +const TICKET_GRACE_MS = 100; +import { TlsSessionStore } from './tls-session-store'; + +// Node.js's networking stack is required lazily, at first use: mongosh loads +// this module (via the @mongosh/logging index) while building its V8 startup +// snapshot, and http/https cannot be included in startup snapshots — the +// snapshot builder aborts with 'CheckGlobalAndEternalHandles failed'. See the +// snapshot handling in packages/cli-repl/src/run.ts. Node caches modules, so +// the lazy require is a map lookup after the first call. +function httpModule(): typeof http { + return require('http') as typeof http; +} +function httpsModule(): typeof https { + return require('https') as typeof https; +} +function tlsModule(): typeof tls { + return require('tls') as typeof tls; +} +function dnsModule(): typeof dns { + return require('dns') as typeof dns; +} + +export type FireAndForgetBeaconOptions = { + /** + * Externally managed agent (e.g. the proxy-aware agent from + * devtools-proxy-support). When set, it is used for both protocols and the + * beacon does not create or destroy agents of its own. + */ + agent?: http.Agent; + /** Headers merged into every request (per-send headers take precedence). */ + defaultHeaders?: Record; + /** Extra TLS options for https connections (e.g. `ca` in tests). */ + tlsOptions?: tls.ConnectionOptions; + /** Adaptive request-timeout tuning; see currentTimeoutMs(). */ + timeouts?: { + /** Timeout until enough samples exist; also the upper bound. Default REQUEST_TIMEOUT_MS. */ + defaultMs?: number; + /** Lower bound for the adaptive timeout. Default 250. */ + minMs?: number; + /** Headroom multiplier over the p90 dispatch duration. Default 4. */ + multiplier?: number; + /** Samples required before the timeout adapts. Default 10. */ + minSamples?: number; + }; + /** Circuit breaker tuning. */ + breaker?: { + /** Consecutive failures before the breaker opens. Default 5. */ + threshold?: number; + /** How long the breaker stays open before allowing one probe. Default 5 minutes. */ + cooldownMs?: number; + }; + /** DNS lookup override; used by the built-in agents. */ + lookup?: typeof dns.lookup; + /** TTL for the built-in DNS cache; default 60s. Ignored when `lookup` is set. */ + dnsCacheTtlMs?: number; + /** + * Path of the persisted TLS session-ticket store. When set (and no external + * `agent` is given), https connections resume sessions across processes. + */ + sessionStorePath?: string; +}; + +/** + * Wraps dns.lookup with a tiny TTL cache so repeat connections to the + * telemetry endpoint skip the DNS round-trip. Multi-answer (`all: true`) + * lookups are passed through uncached. + */ +export function createCachedLookup( + ttlMs: number, + baseLookup?: typeof dns.lookup +): typeof dns.lookup { + const base = baseLookup ?? dnsModule().lookup; + const cache = new Map< + string, + { address: string; family: number; expiresAt: number } + >(); + // eslint-disable-next-line @typescript-eslint/no-explicit-any + return function lookup(hostname: string, options: any, callback?: any): any { + if (typeof options === 'function') { + callback = options; + options = {}; + } + if (options.all) { + // eslint-disable-next-line @typescript-eslint/no-unsafe-argument + return base(hostname, options, callback); + } + // eslint-disable-next-line @typescript-eslint/restrict-template-expressions + const key = `${hostname}|${options.family ?? 0}`; + const hit = cache.get(key); + if (hit && hit.expiresAt > Date.now()) { + callback(null, hit.address, hit.family); + return; + } + base( + hostname, + // eslint-disable-next-line @typescript-eslint/no-unsafe-argument + options, + // eslint-disable-next-line @typescript-eslint/no-explicit-any + (err: NodeJS.ErrnoException | null, address: any, family: any) => { + if (!err && typeof address === 'string') { + cache.set(key, { + address, + family, + expiresAt: Date.now() + ttlMs, + }); + } + callback(err, address, family); + } + ); + } as typeof dns.lookup; +} + +/** + * https.Agent that persists TLS session tickets via TlsSessionStore and + * offers them on new connections, so a fresh process resumes the handshake + * in one round-trip instead of performing a full one. + */ +export interface ResumingHttpsAgent extends https.Agent { + /** Resolves when the first session ticket of this process is captured. */ + readonly firstTicket: Promise; + /** + * True once a handshake has completed but its ticket has not arrived yet. + * TLS 1.3 delivers NewSessionTicket ~1 RTT after the handshake — after + * `dispatched` resolves — so at shutdown this indicates a ticket is still + * worth a bounded wait (see FireAndForgetBeacon.flush()). + * + * Gated on the handshake actually *completing*, not merely being + * attempted: if the endpoint is down or refuses the connection, no + * ticket can ever arrive, and flush() should not pay the grace period in + * exactly the failure case the circuit breaker exists for. + */ + readonly awaitingFirstTicket: boolean; +} + +type ResumingHttpsAgentConstructor = new ( + options: https.AgentOptions, + store: TlsSessionStore +) => ResumingHttpsAgent; + +let ResumingHttpsAgentClass: ResumingHttpsAgentConstructor | undefined; + +// The class is defined lazily: `extends https.Agent` at module scope would +// force the https require this file goes out of its way to defer (see the +// snapshot note on the lazy module helpers above). +function getResumingHttpsAgentClass(): ResumingHttpsAgentConstructor { + if (ResumingHttpsAgentClass) return ResumingHttpsAgentClass; + const { Agent } = httpsModule(); + class ResumingHttpsAgent extends Agent { + private readonly store: TlsSessionStore; + private handshakeCompleted = false; + private ticketCaptured = false; + private notifyFirstTicket?: () => void; + readonly firstTicket: Promise; + + constructor(options: https.AgentOptions, store: TlsSessionStore) { + super(options); + this.store = store; + this.firstTicket = new Promise( + (resolve) => (this.notifyFirstTicket = resolve) + ); + } + + get awaitingFirstTicket(): boolean { + return this.handshakeCompleted && !this.ticketCaptured; + } + + createConnection( + options: net.NetConnectOpts, + callback?: (err: Error | null, stream: Duplex) => void + ): Duplex { + const host = 'host' in options && options.host ? options.host : 'localhost'; + + const connectOptions = { + ...options, + host, + session: this.store.get(host), + }; + + const socket = super.createConnection(connectOptions, callback) as tls.TLSSocket; + + socket.once('secureConnect', () => { + this.handshakeCompleted = true; + if (socket.isSessionReused()) { + // A resumed session proves the stored ticket is still valid, and + // servers do not necessarily issue a fresh NewSessionTicket on + // resumed connections — without this, flush() would wait the + // full ticket grace on every session after the first. + this.ticketCaptured = true; + this.notifyFirstTicket?.(); + } + }); + + // TLS 1.3 delivers session tickets after the handshake + socket.on('session', (ticket: Buffer) => { + this.ticketCaptured = true; + this.notifyFirstTicket?.(); + this.store.set(host, ticket); + }); + + return socket; + } + } + + ResumingHttpsAgentClass = ResumingHttpsAgent; + return ResumingHttpsAgentClass; +} + +/** + * A fire-and-forget HEAD launcher built on the raw http/https modules. + * + * Unlike fetch, `send()` resolves as soon as the request has been fully + * written to an *established* connection ('finish' + 'connect'/'secureConnect') + * instead of waiting for the response: once the bytes reach the kernel, TCP + * delivers them even if the process exits immediately afterwards. Every socket + * is unref'd so pending telemetry can never keep the mongosh process alive. + * + * Keep-alive agents reuse connections across sends when the server responds + * promptly, but maxSockets is deliberately left unbounded: keep-alive reuse + * requires the previous *response* to complete, so a socket cap would queue + * burst sends behind a stalled response — the exact hang this class exists + * to eliminate. + */ +export class FireAndForgetBeacon implements Beacon { + protected readonly options: FireAndForgetBeaconOptions; + private httpAgent?: http.Agent; + private httpsAgent?: https.Agent; + private sessionStore?: TlsSessionStore; + private resumingAgent?: ResumingHttpsAgent; + /** Durations of recent successful dispatches (ring of 50). */ + private readonly dispatchDurations: number[] = []; + private consecutiveFailures = 0; + private breakerOpenedAt?: number; + private probeInFlight = false; + private hasSent = false; + private lookup?: typeof dns.lookup; + + constructor(options: FireAndForgetBeaconOptions = {}) { + this.options = options; + if (options.sessionStorePath) { + // Constructed eagerly so its async disk read starts now — long before + // the first send — instead of racing the first TLS connect, where a + // miss would silently disable session resumption for exactly the + // short-lived sessions it exists for. send() awaits whenLoaded(). + this.sessionStore = new TlsSessionStore(options.sessionStorePath); + } + } + + private agentFor(isHttps: boolean): http.Agent { + if (this.options.agent) return this.options.agent; + this.lookup ??= + this.options.lookup ?? + createCachedLookup(this.options.dnsCacheTtlMs ?? 60_000); + const agentOptions = { keepAlive: true, lookup: this.lookup }; + if (isHttps) { + if (!this.httpsAgent) { + if (this.sessionStore) { + this.resumingAgent = new (getResumingHttpsAgentClass())( + { ...agentOptions, ...this.options.tlsOptions }, + this.sessionStore + ); + this.httpsAgent = this.resumingAgent; + } else { + this.httpsAgent = new (httpsModule().Agent)({ + ...agentOptions, + ...this.options.tlsOptions, + }); + } + } + return this.httpsAgent; + } + this.httpAgent ??= new (httpModule().Agent)(agentOptions); + return this.httpAgent; + } + + async send( + url: string, + headers: Record + ): Promise { + if (this.breakerIsOpen()) { + // The endpoint has been consistently failing (down or firewalled); + // act as /dev/null instead of burning sockets and timeouts. + return { kind: 'suppressed' }; + } + // Wait for the persisted TLS tickets to be readable (async, usually long + // since resolved) so the first connect of a short-lived session can + // still resume instead of racing the disk read and missing. + await this.sessionStore?.whenLoaded(); + const outcome = await this.doSend(url, headers); + return this.recordOutcome(outcome); + } + + /** + * The request timeout the next send will use. Starts at the generous + * default; once enough dispatch durations are observed, tightens to a + * p90-with-headroom so a hanging endpoint is abandoned at a deadline + * scaled to this host's actual network, never a hardcoded worst case. + * (p90, not mean: telemetry RTTs have fat tails, and a mean-derived cap + * would abort legitimate slow sends and feed back into the breaker.) + */ + currentTimeoutMs(): number { + const { + defaultMs = REQUEST_TIMEOUT_MS, + minMs = 250, + multiplier = 4, + minSamples = 10, + } = this.options.timeouts ?? {}; + if (this.dispatchDurations.length < minSamples) return defaultMs; + const sorted = [...this.dispatchDurations].sort((a, b) => a - b); + const p90 = + sorted[Math.min(sorted.length - 1, Math.floor(sorted.length * 0.9))]; + return Math.max(minMs, Math.min(defaultMs, Math.ceil(p90 * multiplier))); + } + + private recordOutcome(outcome: BeaconOutcome): BeaconOutcome { + if (outcome.kind === 'dispatched') { + this.consecutiveFailures = 0; + this.breakerOpenedAt = undefined; + this.probeInFlight = false; + this.dispatchDurations.push(outcome.durationMs); + if (this.dispatchDurations.length > 50) this.dispatchDurations.shift(); + } else if (outcome.kind === 'error') { + const wasProbe = this.probeInFlight; + this.probeInFlight = false; + this.consecutiveFailures++; + const { threshold = 5 } = this.options.breaker ?? {}; + if (wasProbe) { + // Failed probe: reopen immediately, restarting the cooldown. + this.breakerOpenedAt = Date.now(); + } else if ( + this.consecutiveFailures >= threshold && + this.breakerOpenedAt === undefined + ) { + this.breakerOpenedAt = Date.now(); + } + } + return outcome; + } + + private breakerIsOpen(): boolean { + if (this.breakerOpenedAt === undefined) return false; + // A half-open probe is already in flight — stay suppressed until its + // outcome is recorded. + if (this.probeInFlight) return true; + const { cooldownMs = 300_000 } = this.options.breaker ?? {}; + if (Date.now() - this.breakerOpenedAt < cooldownMs) return true; + // Half-open: exactly this send becomes the probe. breakerOpenedAt stays + // set so concurrent sends remain suppressed while the probe is out. + this.probeInFlight = true; + return false; + } + + private doSend( + url: string, + headers: Record + ): Promise { + this.hasSent = true; + return new Promise((resolve) => { + const start = performance.now(); + const done = (outcome: BeaconOutcome): void => resolve(outcome); + + let target: URL; + try { + target = new URL(url); + } catch (error) { + done({ kind: 'error', error: error as Error, durationMs: 0 }); + return; + } + const isHttps = target.protocol === 'https:'; + + let req: http.ClientRequest; + try { + req = (isHttps ? httpsModule() : httpModule()).request(target, { + method: 'HEAD', + headers: { ...this.options.defaultHeaders, ...headers }, + agent: this.agentFor(isHttps), + signal: AbortSignal.timeout(this.currentTimeoutMs()), + ...(isHttps ? this.options.tlsOptions : {}), + }); + } catch (error) { + done({ + kind: 'error', + error: error as Error, + durationMs: performance.now() - start, + }); + return; + } + + let connected = false; + let finished = false; + const maybeDispatched = (): void => { + if (connected && finished) { + done({ kind: 'dispatched', durationMs: performance.now() - start }); + } + }; + + req.on('socket', (socket) => { + // Never let a pending telemetry request keep the process alive. + socket.unref(); + if (!socket.connecting) { + // Reused keep-alive socket — already established. + connected = true; + maybeDispatched(); + return; + } + // 'dispatched' means the bytes reached the kernel; that is only true + // once the connection (including the TLS handshake) is established. + const connectEvent = + socket instanceof tlsModule().TLSSocket ? 'secureConnect' : 'connect'; + socket.once(connectEvent, () => { + connected = true; + maybeDispatched(); + }); + }); + req.on('finish', () => { + finished = true; + maybeDispatched(); + }); + // Drain the response so the keep-alive socket returns to the pool. + // (`done` is a no-op by then — the promise already resolved.) + req.on('response', (res) => res.resume()); + req.on('error', (error) => + done({ kind: 'error', error, durationMs: performance.now() - start }) + ); + req.end(); + }); + } + + /** + * Impending-shutdown hook, invoked via TelemetryClient.flush(). TLS 1.3 + * delivers session tickets ~1 RTT *after* the handshake — i.e. after + * `dispatched` resolves — so a short-lived session that only sends at exit + * would otherwise never seed the resumption cache. Grant a small bounded + * grace for an in-flight ticket, then await the pending store write. + */ + async flush(): Promise { + if (this.hasSent) { + // Half-close every beacon-owned socket now, so a TLS close_notify and + // TCP FIN reach the kernel before the process concludes — otherwise + // exit tears the connections down abruptly (RST, no close_notify) and + // the server logs every session as an error. Half-close still lets + // responses and TLS 1.3 session tickets arrive below. + this.endOwnedSockets(); + // 'dispatched' resolves when the request is written to the TLS/socket + // layer, which is still userspace — and with every telemetry handle + // unref'd, awaiting alone does not keep the event loop alive to drain + // it into the kernel. This short, deliberately ref'd delay is what + // keeps the process running those last few milliseconds (including + // the close_notify/FIN just queued above). Bounded and cleared by + // construction (plain one-shot timer). + await new Promise((resolve) => + setTimeout(resolve, FLUSH_SETTLE_DELAY_MS) + ); + } + if (this.resumingAgent?.awaitingFirstTicket) { + // Ref'd for the same reason as above; cleared as soon as the ticket + // arrives so it only bounds the wait rather than extending it. + let timer: ReturnType | undefined; + const grace = new Promise( + (resolve) => (timer = setTimeout(resolve, TICKET_GRACE_MS)) + ); + try { + await Promise.race([this.resumingAgent.firstTicket, grace]); + } finally { + if (timer !== undefined) clearTimeout(timer); + } + } + await this.sessionStore?.flush(); + } + + /** + * Half-close (FIN + TLS close_notify) every socket of the beacon-owned + * agents. Deliberately does NOT touch an externally provided agent + * (options.agent) — that one is shared with the rest of the shell. + */ + private endOwnedSockets(): void { + for (const agent of [this.httpAgent, this.httpsAgent]) { + if (!agent) continue; + for (const pool of [agent.sockets, agent.freeSockets]) { + for (const sockets of Object.values(pool)) { + for (const socket of sockets ?? []) { + socket.end(); + } + } + } + } + } + + /** + * Fire a HEAD request purely to establish the connection (DNS + TCP + TLS) + * so that the first real event is a bare write on a hot socket. The + * outcome is intentionally ignored. + */ + warmUp(url: string): void { + void this.send(url, {}); + } + + /** Destroys the beacon-owned agents and their pooled sockets. */ + close(): void { + this.httpAgent?.destroy(); + this.httpsAgent?.destroy(); + } +} diff --git a/packages/logging/src/index.ts b/packages/logging/src/index.ts index 3695f5ab45..42e4369d35 100644 --- a/packages/logging/src/index.ts +++ b/packages/logging/src/index.ts @@ -4,7 +4,12 @@ export { NoopAnalytics, ThrottledAnalytics, } from './analytics-helpers'; -export { TelemetryClient, REQUEST_TIMEOUT_MS } from './telemetry-client'; +export { TelemetryClient } from './telemetry-client'; +export { REQUEST_TIMEOUT_MS } from './beacon'; +export type { Beacon, BeaconOutcome } from './beacon'; +export { FireAndForgetBeacon } from './fire-and-forget-beacon'; +export type { FireAndForgetBeaconOptions } from './fire-and-forget-beacon'; +export { TlsSessionStore } from './tls-session-store'; export { MongoshLoggingAndTelemetry } from './types'; export { setupLoggingAndTelemetry } from './logging-and-telemetry'; export { getAiAgent, KNOWN_AGENT_ENV_VARS } from './helpers'; diff --git a/packages/logging/src/telemetry-client.spec.ts b/packages/logging/src/telemetry-client.spec.ts index b93ff09387..1a2365c27e 100644 --- a/packages/logging/src/telemetry-client.spec.ts +++ b/packages/logging/src/telemetry-client.spec.ts @@ -1,7 +1,8 @@ import { expect } from 'chai'; import { gunzipSync } from 'zlib'; import { TelemetryClient } from '.'; -import type { TelemetryEvent } from '.'; +import type { TelemetryEvent, Beacon } from '.'; +import type { BeaconOutcome } from './beacon'; const sessionEvent: TelemetryEvent = { name: 'Identify', @@ -25,41 +26,47 @@ const sessionEvent: TelemetryEvent = { }, }; +type RecordedSend = { url: string; headers: Record }; + +function createFakeBeacon( + sendImpl?: ( + url: string, + headers: Record + ) => Promise +): { beacon: Beacon; sends: RecordedSend[] } { + const sends: RecordedSend[] = []; + const beacon: Beacon = { + send(url, headers) { + sends.push({ url, headers }); + return ( + sendImpl?.(url, headers) ?? + Promise.resolve({ kind: 'dispatched', durationMs: 1 }) + ); + }, + }; + return { beacon, sends }; +} + describe('TelemetryClient', function () { - it('sends events to the configured endpoint', async function () { - const calls: string[] = []; - const client = new TelemetryClient('https://example.com/events', (url) => { - calls.push(url); - return Promise.resolve(); - }); + it('send events to the configured endpoint', async function () { + const { beacon, sends } = createFakeBeacon(); + const client = new TelemetryClient('https://example.com/events', beacon); client.track(sessionEvent); await client.flush(); - expect(calls).to.deep.equal([ + expect(sends.map(({ url }) => url)).to.deep.equal([ 'https://example.com/events/v1/identify?deviceId=test-device&sessionId=test-session', ]); }); - it('sends a HEAD request with the event gzip+base64-encoded in the Cookie header', async function () { - const requests: { url: string; init: any }[] = []; - const client = new TelemetryClient( - 'https://example.com/events', - (url, init) => { - requests.push({ url, init }); - return Promise.resolve(); - } - ); + it('send the event gzip+base64-encoded in the Cookie header', async function () { + const { beacon, sends } = createFakeBeacon(); + const client = new TelemetryClient('https://example.com/events', beacon); client.track(sessionEvent); await client.flush(); - expect(requests).to.have.lengthOf(1); - expect(requests[0].url).to.equal( - 'https://example.com/events/v1/identify?deviceId=test-device&sessionId=test-session' - ); - expect(requests[0].init.method).to.equal('HEAD'); - expect(requests[0].init.signal).to.be.instanceOf(AbortSignal); - - const cookie: string = requests[0].init.headers.Cookie; + expect(sends).to.have.lengthOf(1); + const cookie: string = sends[0].headers.Cookie; expect(cookie).to.match(/^mge=/); const decoded = gunzipSync( Buffer.from(cookie.slice('mge='.length), 'base64') @@ -69,59 +76,32 @@ describe('TelemetryClient', function () { ); }); - it('aborts a request that never resolves after the request timeout', async function () { - let capturedSignal: AbortSignal | undefined; - const client = new TelemetryClient( - 'https://example.com/events', - (_url, init) => { - capturedSignal = init?.signal; - // Simulate a stuck network request and reject once the signal is aborted. - return new Promise((_resolve, reject) => { - init?.signal?.addEventListener('abort', () => { - reject(new Error('The operation was aborted')); - }); - }); - }, - undefined, - 5 // requestTimeoutMs + it('stay silent when the beacon rejects despite its contract', async function () { + const { beacon } = createFakeBeacon(() => + Promise.reject(new Error('beacon contract violation')) ); - - client.track(sessionEvent); - - // The request never resolves on its own. flush() only returns once the - // timeout aborts it (within the 2s). - await client.flush(); - - expect(capturedSignal).to.be.instanceOf(AbortSignal); - expect(capturedSignal?.aborted).to.equal(true); - }); - - it('silently ignores network errors', async function () { - const client = new TelemetryClient('https://example.com/events', () => { - return Promise.reject(new Error('network failure')); - }); + const client = new TelemetryClient('https://example.com/events', beacon); client.track(sessionEvent); await client.flush(); // must not throw }); - it('flush() resolves immediately when no events were tracked', async function () { - const client = new TelemetryClient('https://example.com/events', () => - Promise.resolve() - ); + it('resolve flush() immediately when no events were tracked', async function () { + const { beacon } = createFakeBeacon(); + const client = new TelemetryClient('https://example.com/events', beacon); await client.flush(); // must not throw }); - it('flush() waits for all in-flight requests before resolving', async function () { - let resolve1!: () => void; - let resolve2!: () => void; - const p1 = new Promise((r) => (resolve1 = r)); - const p2 = new Promise((r) => (resolve2 = r)); - const responses = [p1, p2]; - let responseIndex = 0; - - const client = new TelemetryClient('https://example.com/events', () => { - return responses[responseIndex++]; - }); + it('wait for all in-flight sends before resolving flush()', async function () { + const dispatched: BeaconOutcome = { kind: 'dispatched', durationMs: 1 }; + let resolve1!: (o: BeaconOutcome) => void; + let resolve2!: (o: BeaconOutcome) => void; + const outcomes = [ + new Promise((r) => (resolve1 = r)), + new Promise((r) => (resolve2 = r)), + ]; + let sendIndex = 0; + const { beacon } = createFakeBeacon(() => outcomes[sendIndex++]); + const client = new TelemetryClient('https://example.com/events', beacon); client.track(sessionEvent); client.track(sessionEvent); @@ -131,38 +111,37 @@ describe('TelemetryClient', function () { flushed = true; }); - await Promise.resolve(); + await new Promise(setImmediate); expect(flushed).to.equal(false); - resolve1(); - await Promise.resolve(); + resolve1(dispatched); + await new Promise(setImmediate); expect(flushed).to.equal(false); - resolve2(); + resolve2(dispatched); await flushPromise; expect(flushed).to.equal(true); }); - it('flush() clears inflight so a second flush() has nothing to wait on', async function () { - const fetchCalls: number[] = []; - const client = new TelemetryClient('https://example.com/events', () => { - fetchCalls.push(1); - return Promise.resolve(); - }); + it('clear inflight so a second flush() has nothing to wait on', async function () { + const { beacon, sends } = createFakeBeacon(); + const client = new TelemetryClient('https://example.com/events', beacon); client.track(sessionEvent); await client.flush(); - expect(fetchCalls).to.have.lengthOf(1); + expect(sends).to.have.lengthOf(1); await client.flush(); // no new track() calls — should resolve immediately - expect(fetchCalls).to.have.lengthOf(1); + expect(sends).to.have.lengthOf(1); }); - it('flush() resolves via timeout when a request never completes', async function () { - // Simulate a stuck network request that never resolves. + it('resolve flush() via the timeout when a send never completes', async function () { + const { beacon } = createFakeBeacon( + () => new Promise(() => undefined) // Never resolves. + ); const client = new TelemetryClient( 'https://example.com/events', - () => new Promise(() => undefined), // Never resolves. + beacon, 10 // Override the 2s default so the test completes much faster. ); client.track(sessionEvent); @@ -171,28 +150,98 @@ describe('TelemetryClient', function () { expect(Date.now() - start).to.be.lessThan(500); // Well within CI tolerance. }); - it('events tracked after flush() starts are not included in that flush', async function () { - let resolveFirst!: () => void; - const firstDone = new Promise((r) => (resolveFirst = r)); - let fetchCount = 0; - - const client = new TelemetryClient('https://example.com/events', () => { - fetchCount++; - if (fetchCount === 1) return firstDone; - return Promise.resolve(); + it('exclude events tracked after flush() starts from that flush', async function () { + const dispatched: BeaconOutcome = { kind: 'dispatched', durationMs: 1 }; + let resolveFirst!: (o: BeaconOutcome) => void; + const firstOutcome = new Promise((r) => (resolveFirst = r)); + let sendCount = 0; + const { beacon } = createFakeBeacon(() => { + sendCount++; + if (sendCount === 1) return firstOutcome; + return Promise.resolve(dispatched); }); + const client = new TelemetryClient('https://example.com/events', beacon); client.track(sessionEvent); // first event — held until resolveFirst() const flushPromise = client.flush(); client.track(sessionEvent); // second event tracked while flush is pending - resolveFirst(); + resolveFirst(dispatched); await flushPromise; // second event is in a fresh inflight batch, not in the completed flush; // draining it separately confirms it was tracked outside the first flush await client.flush(); // drain the second event - expect(fetchCount).to.equal(2); + expect(sendCount).to.equal(2); + }); + + it('invoke the beacon flush hook after in-flight sends complete', async function () { + const order: string[] = []; + const beacon: Beacon = { + send: async () => { + await new Promise((resolve) => setTimeout(resolve, 20)); + order.push('send'); + return { kind: 'dispatched', durationMs: 1 }; + }, + flush: () => { + order.push('beacon-flush'); + return Promise.resolve(); + }, + }; + const client = new TelemetryClient('https://example.com/events', beacon); + client.track(sessionEvent); + client.track(sessionEvent); + await client.flush(); + expect(order).to.deep.equal(['send', 'send', 'beacon-flush']); + }); + + it('call the beacon flush hook even when no events were tracked', async function () { + let flushCalls = 0; + const beacon: Beacon = { + send: () => Promise.resolve({ kind: 'dispatched', durationMs: 1 }), + flush: () => { + flushCalls++; + return Promise.resolve(); + }, + }; + const client = new TelemetryClient('https://example.com/events', beacon); + await client.flush(); + expect(flushCalls).to.equal(1); + }); + + it('bound the beacon flush hook by the flush timeout', async function () { + const beacon: Beacon = { + send: () => Promise.resolve({ kind: 'dispatched', durationMs: 1 }), + flush: () => new Promise(() => undefined), // never resolves + }; + const client = new TelemetryClient( + 'https://example.com/events', + beacon, + 10 + ); + client.track(sessionEvent); + const start = Date.now(); + await client.flush(); + expect(Date.now() - start).to.be.lessThan(500); + }); + + it('forward warm-up to the beacon with the /warm-up path', function () { + const warmUpCalls: string[] = []; + const beacon: Beacon = { + send: () => Promise.resolve({ kind: 'dispatched', durationMs: 1 }), + warmUp: (url) => { + warmUpCalls.push(url); + }, + }; + const client = new TelemetryClient('https://example.com/events', beacon); + client.warmUp(); + expect(warmUpCalls).to.deep.equal(['https://example.com/events/warm-up']); + }); + + it('tolerate beacons without warm-up support', function () { + const { beacon } = createFakeBeacon(); + const client = new TelemetryClient('https://example.com/events', beacon); + client.warmUp(); // must not throw }); }); diff --git a/packages/logging/src/telemetry-client.ts b/packages/logging/src/telemetry-client.ts index 305847183a..4bd145c782 100644 --- a/packages/logging/src/telemetry-client.ts +++ b/packages/logging/src/telemetry-client.ts @@ -2,45 +2,36 @@ import { gzip } from 'zlib'; import { promisify } from 'util'; import type { TelemetryEvent } from './telemetry-events'; import type { MongoshAnalytics } from './analytics-helpers'; +import type { Beacon } from './beacon'; const gzipAsync = promisify(gzip); -// Generous enough number for a fire-and-forget event. Adjust freely if needed. -export const REQUEST_TIMEOUT_MS = 5_000; - const FLUSH_TIMEOUT_MS = 2_000; const SCHEMA_VERSION = 'v1'; -type FetchFn = ( - url: string, - init?: { - method?: string; - headers?: { Cookie: string }; - signal?: AbortSignal; - } -) => Promise; - function eventPath(name: TelemetryEvent['name']): string { return `/${SCHEMA_VERSION}/${name.toLowerCase().replace(/\s+/g, '-')}`; } export class TelemetryClient implements MongoshAnalytics { private readonly endpoint: string; - private readonly fetch: FetchFn; + private readonly beacon: Beacon; private readonly flushTimeoutMs: number; - private readonly requestTimeoutMs: number; private readonly inflight: Promise[] = []; constructor( endpoint: string, - fetch: FetchFn = globalThis.fetch.bind(globalThis), - flushTimeoutMs: number = FLUSH_TIMEOUT_MS, - requestTimeoutMs: number = REQUEST_TIMEOUT_MS + beacon: Beacon, + flushTimeoutMs: number = FLUSH_TIMEOUT_MS ) { this.endpoint = endpoint; - this.fetch = fetch; + this.beacon = beacon; this.flushTimeoutMs = flushTimeoutMs; - this.requestTimeoutMs = requestTimeoutMs; + } + + /** Pre-establish the connection to the telemetry endpoint, if the beacon supports it. */ + warmUp(): void { + this.beacon.warmUp?.(`${this.endpoint}/warm-up`); } /** @@ -48,7 +39,7 @@ export class TelemetryClient implements MongoshAnalytics { * - path (cs-uri-stem): schema version + event name, e.g. /v1/new-connection * - query string (cs-uri-query): device_id / session_id, for filtering & joins in raw logs * - User-Agent (cs(User-Agent)): client identity (mongosh version, OS, arch), - * attached by the `fetch` passed into the + * attached by the Beacon passed into the * constructor, not by this class * - Cookie (cs(Cookie)): full event payload, gzip-compressed + base64-encoded * @@ -66,28 +57,47 @@ export class TelemetryClient implements MongoshAnalytics { // and/or use a custom dictionary rather than plain gzip. const p = gzipAsync(Buffer.from(JSON.stringify(event))) .then((compressed) => - this.fetch(url, { - method: 'HEAD', - headers: { Cookie: `mge=${compressed.toString('base64')}` }, - signal: AbortSignal.timeout(this.requestTimeoutMs), + this.beacon.send(url, { + Cookie: `mge=${compressed.toString('base64')}`, }) ) .then(() => { - // discard the Response; callers only await completion + // discard the outcome; callers only await completion }) .catch(() => { - // telemetry is best-effort; ignore send failures (including timeouts) + // telemetry is best-effort; guards gzip/serialization failures and + // beacons that reject despite their contract }); this.inflight.push(p); } - // TODO(MONGOSH-3454): Optimize aggregated event flushing. + /** + * Bounded shutdown window: waits for in-flight sends to reach the kernel, + * then gives the beacon its impending-shutdown hook (persistence I/O) — + * all raced against flushTimeoutMs so exit can never hang on telemetry. + */ async flush(): Promise { const pending = this.inflight.splice(0); - if (pending.length === 0) return; - const timeout = new Promise((resolve) => - setTimeout(resolve, this.flushTimeoutMs).unref?.() + if (pending.length === 0 && !this.beacon.flush) return; + const work = Promise.all(pending) + .then(() => this.beacon.flush?.()) + .catch(() => { + // the beacon contract never rejects; guard against violations anyway + }); + // This timer is deliberately ref'd: by design every telemetry socket and + // internal timer is unref'd, so during the exit flush nothing else may be + // keeping the event loop alive — awaiting a promise does not. Without a + // ref'd handle the process can exit mid-flush and drop the final events + // while they sit in userspace TLS buffers. The timer is cleared as soon + // as the work settles, so it never delays exit; it only bounds it. + let timer: ReturnType | undefined; + const timeout = new Promise( + (resolve) => (timer = setTimeout(resolve, this.flushTimeoutMs)) ); - await Promise.race([Promise.all(pending), timeout]); + try { + await Promise.race([work, timeout]); + } finally { + if (timer !== undefined) clearTimeout(timer); + } } } diff --git a/packages/logging/src/tls-session-store.spec.ts b/packages/logging/src/tls-session-store.spec.ts new file mode 100644 index 0000000000..0cfe14af0d --- /dev/null +++ b/packages/logging/src/tls-session-store.spec.ts @@ -0,0 +1,91 @@ +import { expect } from 'chai'; +import fs from 'fs'; +import os from 'os'; +import path from 'path'; +import { TlsSessionStore } from './tls-session-store'; + +describe('TlsSessionStore', function () { + let dir: string; + let filePath: string; + + beforeEach(function () { + dir = fs.mkdtempSync(path.join(os.tmpdir(), 'tls-session-store-')); + filePath = path.join(dir, 'sessions.json'); + }); + + afterEach(function () { + fs.rmSync(dir, { recursive: true, force: true }); + }); + + it('round-trip a session ticket through the file', async function () { + const ticket = Buffer.from('fake-session-ticket'); + const writer = new TlsSessionStore(filePath); + writer.set('telemetry.example.com', ticket); + await writer.flush(); // set() persists asynchronously; flush awaits the write + expect(fs.existsSync(filePath)).to.equal(true); + + const reader = new TlsSessionStore(filePath); // fresh instance = fresh process + await reader.whenLoaded(); // the disk state is read asynchronously + expect(reader.get('telemetry.example.com')).to.deep.equal(ticket); + expect(reader.get('other.example.com')).to.equal(undefined); + }); + + it('expire tickets past the TTL', async function () { + const writer = new TlsSessionStore(filePath, -1); // everything is expired + writer.set('telemetry.example.com', Buffer.from('stale')); + await writer.flush(); + + const reader = new TlsSessionStore(filePath, -1); + await reader.whenLoaded(); + expect(reader.get('telemetry.example.com')).to.equal(undefined); + }); + + it('ignore a corrupted store file silently', async function () { + fs.writeFileSync(filePath, 'not json at all{{{'); + const store = new TlsSessionStore(filePath); + await store.whenLoaded(); // must not reject + expect(store.get('telemetry.example.com')).to.equal(undefined); + store.set('telemetry.example.com', Buffer.from('recovered')); // must not throw + }); + + it('ignore a store file containing non-object JSON silently', async function () { + fs.writeFileSync(filePath, '42'); // valid JSON, wrong shape + const store = new TlsSessionStore(filePath); + await store.whenLoaded(); + expect(store.get('telemetry.example.com')).to.equal(undefined); + store.set('telemetry.example.com', Buffer.from('recovered')); // must not throw + }); + + it('preserve a ticket set before the disk state finishes loading', async function () { + const seed = new TlsSessionStore(filePath); + seed.set('telemetry.example.com', Buffer.from('older-from-disk')); + seed.set('other.example.com', Buffer.from('disk-only')); + await seed.flush(); + + const store = new TlsSessionStore(filePath); + // set() while the async read is still in flight: the newer in-memory + // ticket must survive the load completing, while disk-only entries + // still merge in underneath. + store.set('telemetry.example.com', Buffer.from('newer-in-memory')); + await store.whenLoaded(); + expect(store.get('telemetry.example.com')).to.deep.equal( + Buffer.from('newer-in-memory') + ); + expect(store.get('other.example.com')).to.deep.equal( + Buffer.from('disk-only') + ); + }); + + it('resolve flush() without a pending write', async function () { + const store = new TlsSessionStore(filePath); + await store.flush(); // must not throw or hang + }); + + it('restrict the store file permissions to the owner', async function () { + if (process.platform === 'win32') return this.skip(); + const store = new TlsSessionStore(filePath); + store.set('telemetry.example.com', Buffer.from('secret')); + await store.flush(); + expect(fs.statSync(filePath).mode & 0o777).to.equal(0o600); + }); +}); diff --git a/packages/logging/src/tls-session-store.ts b/packages/logging/src/tls-session-store.ts new file mode 100644 index 0000000000..4a46a8fcbd --- /dev/null +++ b/packages/logging/src/tls-session-store.ts @@ -0,0 +1,101 @@ +import fs from 'fs'; +import path from 'path'; + +type StoredSessions = Record; + +const DEFAULT_TTL_MS = 6 * 3_600_000; // session tickets go stale server-side + +/** + * Persists TLS session tickets across mongosh sessions so the next process + * can resume the telemetry TLS handshake in a single round-trip + * (MONGOSH-3454). Tickets are resumption secrets: the file is written with + * mode 0600 next to mongosh's other local state. This is a cache — every + * failure (missing file, corrupt JSON, failed write) is silent. + * + * The on-disk state is read asynchronously, starting at construction: + * mongosh's startup path is strictly async-fs (a sync read here would be the + * only event-loop-blocking file access in the whole boot sequence, at its + * most concurrent moment). Callers that depend on the persisted tickets + * being visible must await {@link whenLoaded} first; `get()` before that + * simply misses, which for a resumption cache is safe. + */ +export class TlsSessionStore { + // tsconfig has erasableSyntaxOnly: true, which disallows TS parameter + // properties (they desugar to a constructor body assignment, which isn't + // erasable syntax); declared as fields and assigned in the constructor + // body instead. + private readonly filePath: string; + private readonly ttlMs: number; + private sessions?: StoredSessions; + private pendingWrite: Promise = Promise.resolve(); + private readonly loaded: Promise; + + constructor(filePath: string, ttlMs: number = DEFAULT_TTL_MS) { + this.filePath = filePath; + this.ttlMs = ttlMs; + this.loaded = fs.promises + .readFile(this.filePath, 'utf8') + .then((raw) => { + const parsed: unknown = JSON.parse(raw); + const diskSessions = + typeof parsed === 'object' && + parsed !== null && + !Array.isArray(parsed) + ? (parsed as StoredSessions) + : {}; + // A ticket set() while the read was in flight is newer than anything + // on disk — merge under, never clobber. + this.sessions = { ...diskSessions, ...(this.sessions ?? {}) }; + }) + .catch(() => { + this.sessions ??= {}; + }); + } + + /** + * Resolves once the on-disk state has been read (or failed silently). + * Never rejects. + */ + whenLoaded(): Promise { + return this.loaded; + } + + get(host: string): Buffer | undefined { + // Memory-only: before whenLoaded() resolves this misses, which callers + // avoid by awaiting whenLoaded() first (see FireAndForgetBeacon.send). + const entry = this.sessions?.[host]; + if (!entry?.ticket || Date.now() - entry.storedAt > this.ttlMs) { + return undefined; + } + return Buffer.from(entry.ticket, 'base64'); + } + + set(host: string, ticket: Buffer): void { + this.sessions ??= {}; + this.sessions[host] = { + ticket: ticket.toString('base64'), + storedAt: Date.now(), + }; + // Best-effort async persist; within this process the in-memory copy is + // already up to date even if the write never lands. Chained onto the + // previous pendingWrite (rather than replacing it) so back-to-back + // tickets — TLS 1.3 servers commonly send two NewSessionTickets in a + // row — persist in call order instead of racing two unordered + // writeFile calls to the same path, where the older write could win. + this.pendingWrite = this.pendingWrite + .then(() => + fs.promises.mkdir(path.dirname(this.filePath), { recursive: true }) + ) + .then(() => + fs.promises.writeFile(this.filePath, JSON.stringify(this.sessions), { + mode: 0o600, + }) + ) + .catch(() => undefined); + } + + /** Resolves once the most recent persist has landed (or failed silently). */ + flush(): Promise { + return this.pendingWrite; + } +} diff --git a/packages/logging/test/fixtures/beacon-exit-fixture.ts b/packages/logging/test/fixtures/beacon-exit-fixture.ts new file mode 100644 index 0000000000..ada633c7ca --- /dev/null +++ b/packages/logging/test/fixtures/beacon-exit-fixture.ts @@ -0,0 +1,19 @@ +/** + * Fixture for the "let the process exit while the server is still holding the + * request open" test. Sends one beacon to a server that never responds, prints + * the outcome, and then simply reaches the end of the script. Exit must happen + * naturally: unref'd sockets must not keep the event loop alive. + */ +import { FireAndForgetBeacon } from '../../src/fire-and-forget-beacon'; + +const port = Number.parseInt(process.argv[2], 10); + +async function main(): Promise { + const beacon = new FireAndForgetBeacon(); + const outcome = await beacon.send(`http://localhost:${port}/v1/exit-test`, { + Cookie: 'mge=exit-test', + }); + process.stdout.write(JSON.stringify(outcome)); +} + +void main();