fix: keep handler errors from crashing the app

README.md

@@ -39,6 +39,11 @@ Tracing is activated at most once per process. Calling `init()` a second time
 is a no-op, even with different options. There is no way to stop tracing once
 started. Invalid options throw.
 
+microtel catches errors in its own handlers, so a bug in microtel can't crash
+the app it traces. Each error is reported as an `OTelInstrumentationWarning`
+process warning, at most once every 30 seconds with a running count. The
+request it happened on carries on untraced or partly traced.
+
 ### Options
 
 | Option           | Type   | Default                      | Description |

lib/flush.js

@@ -5,6 +5,8 @@ const http = require('node:http');
 const https = require('node:https');
 const { clearTimeout, setImmediate, setTimeout } = require('node:timers');
 
+const { guard } = require('./guard');
+
 const {
   getEndpoint,
   getExportHeaders,
@@ -163,13 +165,10 @@ function addSpan(span) {
     // filled the buffer.
     if (!flushScheduled) {
       flushScheduled = true;
-      setImmediate(() => {
-        flushScheduled = false;
-        flush();
-      });
+      setImmediate(guardedDeferredFlush);
     }
   } else if (flushTimer == null) {
-    flushTimer = setTimeout(flush, flushIntervalMs);
+    flushTimer = setTimeout(guardedTimerFlush, flushIntervalMs);
     flushTimer.unref();
   }
 }
@@ -239,6 +238,15 @@ function flush() {
   return settleAll(pending);
 }
 
+function deferredFlush() {
+  flushScheduled = false;
+  flush();
+}
+
+const guardedDeferredFlush = guard('deferred flush', deferredFlush);
+const guardedTimerFlush = guard('flush timer', flush);
+const guardedBeforeExitFlush = guard('beforeExit flush', flush);
+
 // Counts failures and emits an OTelExportWarning, throttled to at most one
 // per kWarningThrottleMs.
 function warnExportFailure(message) {
@@ -258,7 +266,9 @@ function sendToCollector(body) {
 
   // Settled by done() from whichever comes first: the response ending, a
   // request or response error, or the export deadline. Settling twice is
-  // harmless, since a promise keeps its first resolution.
+  // harmless, since a promise keeps its first resolution. Each handler
+  // settles before anything that can throw, so a failure its guard catches
+  // cannot leave a flush() waiting.
   let settle;
   const exportDone = new Promise((resolve) => {
     settle = resolve;
@@ -295,48 +305,62 @@ function sendToCollector(body) {
           'content-length': encoded.length,
         },
       },
-      (res) => {
+      guard('export response', (res) => {
         // TODO(bengl): Once retry logic is added, parse the response body for
         // ExportTraceServiceResponse.partial_success.rejected_spans.
         res.resume();
-        res.on('end', () => {
-          clearDeadline();
-          // Anything outside 2xx is a failed export: redirects are not
-          // followed, so a 3xx would silently drop every span.
-          if (res.statusCode < 200 || res.statusCode >= 300) {
+        res.on(
+          'end',
+          guard('export response end', () => {
+            clearDeadline();
+            done();
+            // Anything outside 2xx is a failed export: redirects are not
+            // followed, so a 3xx would silently drop every span.
+            if (res.statusCode < 200 || res.statusCode >= 300) {
+              warnExportFailure(
+                `OTLP collector responded with HTTP ${res.statusCode}`,
+              );
+            }
+          }),
+        );
+        res.on(
+          'error',
+          guard('export response error', (err) => {
+            clearDeadline();
+            done();
             warnExportFailure(
-              `OTLP collector responded with HTTP ${res.statusCode}`,
+              `OTLP export response stream error: ${err.message}`,
             );
-          }
-          done();
-        });
-        res.on('error', (err) => {
-          clearDeadline();
-          warnExportFailure(
-            `OTLP export response stream error: ${err.message}`,
-          );
-          done();
-        });
-      },
+          }),
+        );
+      }),
     );
 
     // exportTimeout is an absolute deadline for the whole export, not a
     // socket idle timeout: a collector that trickles response bytes would
-    // reset an idle timer forever.
-    const deadline = setTimeout(() => {
-      req.destroy(
-        new Error(`exceeded the ${exportTimeoutMs}ms export timeout`),
-      );
-    }, exportTimeoutMs);
+    // reset an idle timer forever. The deadline settles the export itself,
+    // so a flush() waiting on it resolves by then, whatever else fails.
+    const deadline = setTimeout(
+      guard('export timeout', () => {
+        done();
+        req.destroy(
+          new Error(`exceeded the ${exportTimeoutMs}ms export timeout`),
+        );
+      }),
+      exportTimeoutMs,
+    );
     const clearDeadline = () => clearTimeout(deadline);
 
-    req.on('error', (err) => {
-      clearDeadline();
-      warnExportFailure(
-        `Failed to export spans to ${endpoint}: ${err.message}`,
-      );
-      done();
-    });
+    req.on(
+      'error',
+      guard('export request error', (err) => {
+        clearDeadline();
+        done();
+        warnExportFailure(
+          `Failed to export spans to ${endpoint}: ${err.message}`,
+        );
+      }),
+    );
 
     req.end(encoded);
   } catch (err) {
@@ -360,7 +384,7 @@ function startFlusher(options) {
   flushIntervalMs = options?.flushInterval ?? kDefaultFlushIntervalMs;
   exportTimeoutMs = options?.exportTimeout ?? kDefaultExportTimeoutMs;
 
-  process.on('beforeExit', flush);
+  process.on('beforeExit', guardedBeforeExitFlush);
 }
 
 // Resets per-run state for the test suite; tracing itself never resets

lib/guard.js

@@ -0,0 +1,46 @@
+'use strict';
+
+// Node rethrows an error from a diagnostics_channel subscriber or a timer
+// as an uncaught exception, and one from an event listener skips the
+// listeners after it. Every function microtel hands to Node goes through
+// guard(), so a bug in microtel can't do either to the app it traces.
+
+const kWarningThrottleMs = 30_000;
+
+let failureCount = 0;
+let lastWarningTime = 0;
+
+// The warnings are throttled like export failures, so a bug that fires
+// on every request can't flood the logs.
+function warnInstrumentationFailure(name, err) {
+  try {
+    failureCount++;
+    const now = Date.now();
+    if (now - lastWarningTime < kWarningThrottleMs) return;
+    lastWarningTime = now;
+    const suffix = failureCount > 1 ? ` (${failureCount} total failures)` : '';
+    process.emitWarning(
+      `microtel ${name} handler failed: ${err?.message ?? err}${suffix}`,
+      { type: 'OTelInstrumentationWarning', detail: err?.stack },
+    );
+  } catch {
+    // Reporting a failure must not become a new failure.
+  }
+}
+
+// Each wrapper is made once per handler, so a guarded call allocates
+// nothing. A plain function keeps `this` for the response listeners.
+function guard(name, handler) {
+  return function () {
+    try {
+      handler.apply(this, arguments);
+    } catch (err) {
+      warnInstrumentationFailure(name, err);
+    }
+  };
+}
+
+module.exports = {
+  guard,
+  warnInstrumentationFailure,
+};

lib/instrumentations.js

@@ -2,6 +2,8 @@
 
 const dc = require('node:diagnostics_channel');
 
+const { guard } = require('./guard');
+
 const {
   Span,
   SPAN_KIND_SERVER,
@@ -104,7 +106,7 @@ function onHttpServerRequestStart({ request, response, socket }) {
   // before the response completed. Request 'close' cannot be used for
   // this: it also fires when the request stream is merely consumed, which
   // can happen long before a delayed response is sent.
-  response.on('close', onServerResponseClose);
+  response.on('close', guardedServerResponseClose);
 }
 
 // Shared by every server response, so no closure is allocated per
@@ -118,6 +120,11 @@ function onServerResponseClose() {
   }
 }
 
+const guardedServerResponseClose = guard(
+  'server response close',
+  onServerResponseClose,
+);
+
 function onHttpServerResponseFinish({ response }) {
   const span = response[kSpan];
   if (span == null) return;
@@ -216,8 +223,8 @@ function onHttpClientResponseFinish({ request, response }) {
   // covers aborted responses. Ending a span twice is a no-op, so plain
   // listeners do what once() would without its wrappers.
   response[kSpan] = span;
-  response.on('end', endClientResponseSpan);
-  response.on('close', endClientResponseSpan);
+  response.on('end', guardedClientResponseSpanEnd);
+  response.on('close', guardedClientResponseSpanEnd);
 }
 
 // Shared by every client response, so no closure is allocated per
@@ -227,6 +234,11 @@ function endClientResponseSpan() {
   this[kSpan].end();
 }
 
+const guardedClientResponseSpanEnd = guard(
+  'client response end',
+  endClientResponseSpan,
+);
+
 function onClientRequestError({ request, error }) {
   const span = request[kSpan];
   if (span == null) return;
@@ -424,37 +436,39 @@ function onUndiciRequestTrailers({ request }) {
   span.end();
 }
 
+const subscribe = (name, handler) => dc.subscribe(name, guard(name, handler));
+
 function enableInstrumentations() {
   spanStorage = getSpanStorage();
 
   if (isModuleEnabled('node:http/server')) {
-    dc.subscribe('http.server.request.start', onHttpServerRequestStart);
-    dc.subscribe('http.server.response.finish', onHttpServerResponseFinish);
+    subscribe('http.server.request.start', onHttpServerRequestStart);
+    subscribe('http.server.response.finish', onHttpServerResponseFinish);
   }
 
   if (isModuleEnabled('node:http/client')) {
-    dc.subscribe('http.client.request.created', onHttpClientRequestCreated);
-    dc.subscribe('http.client.response.finish', onHttpClientResponseFinish);
-    dc.subscribe('http.client.request.error', onClientRequestError);
+    subscribe('http.client.request.created', onHttpClientRequestCreated);
+    subscribe('http.client.response.finish', onHttpClientResponseFinish);
+    subscribe('http.client.request.error', onClientRequestError);
   }
 
   // A filter of node:http2 covers node:http2/server through the prefix
   // match in isModuleEnabled, the same way node:http covers its client and
   // server subscopes.
   if (isModuleEnabled('node:http2/server')) {
-    dc.subscribe('http2.server.stream.start', onHttp2ServerStreamStart);
-    dc.subscribe('http2.server.stream.finish', onHttp2ServerStreamFinish);
-    dc.subscribe('http2.server.stream.close', onHttp2ServerStreamClose);
-    dc.subscribe('http2.server.stream.error', onHttp2ServerStreamError);
+    subscribe('http2.server.stream.start', onHttp2ServerStreamStart);
+    subscribe('http2.server.stream.finish', onHttp2ServerStreamFinish);
+    subscribe('http2.server.stream.close', onHttp2ServerStreamClose);
+    subscribe('http2.server.stream.error', onHttp2ServerStreamError);
   }
 
   // node:fetch is an alias for node:undici: both enable the same
   // channels, which cover every undici request, fetch() included.
   if (isModuleEnabled('node:undici') || isModuleEnabled('node:fetch')) {
-    dc.subscribe('undici:request:create', onUndiciRequestCreate);
-    dc.subscribe('undici:request:headers', onUndiciRequestHeaders);
-    dc.subscribe('undici:request:trailers', onUndiciRequestTrailers);
-    dc.subscribe('undici:request:error', onClientRequestError);
+    subscribe('undici:request:create', onUndiciRequestCreate);
+    subscribe('undici:request:headers', onUndiciRequestHeaders);
+    subscribe('undici:request:trailers', onUndiciRequestTrailers);
+    subscribe('undici:request:error', onClientRequestError);
   }
 }