fix: log connections that timeout, abort, or send unparable data

Author: itslenny

src/http/plugins/log-request.ts

@@ -1,8 +1,13 @@
+import { Socket } from 'net'
 import fastifyPlugin from 'fastify-plugin'
 import { logSchema, redactQueryParamFromRequest } from '@internal/monitoring'
 import { trace } from '@opentelemetry/api'
 import { FastifyRequest } from 'fastify/types/request'
 import { FastifyReply } from 'fastify/types/reply'
+import { IncomingMessage } from 'http'
+import { getConfig } from '../../config'
+import { parsePartialHttp, PartialHttpData } from '@internal/http'
+import { FastifyInstance } from 'fastify'
 
 interface RequestLoggerOptions {
   excludeUrls?: string[]
@@ -22,13 +27,65 @@ declare module 'fastify' {
   }
 }
 
+const { version } = getConfig()
+
 /**
  * Request logger plugin
  * @param options
  */
 export const logRequest = (options: RequestLoggerOptions) =>
   fastifyPlugin(
     async (fastify) => {
+      // Watch for connections that timeout or disconnect before complete HTTP headers are received
+      // Log if socket closes before request is triggered
+      fastify.server.on('connection', (socket) => {
+        let hasRequest = false
+        const startTime = Date.now()
+        const partialData: Buffer[] = []
+        // Only store 2kb per request
+        const captureByteLimit = 2048
+
+        // Capture partial data sent before connection closes
+        const onData = (chunk: Buffer) => {
+          const remaining = captureByteLimit - partialData.length
+          if (remaining > 0) {
+            partialData.push(chunk.subarray(0, Math.min(chunk.length, remaining)))
+          }
+
+          if (partialData.length >= captureByteLimit) {
+            socket.removeListener('data', onData)
+          }
+        }
+        socket.on('data', onData)
+
+        // Track if this socket ever receives an HTTP request
+        const onRequest = (req: IncomingMessage) => {
+          if (req.socket === socket) {
+            hasRequest = true
+            socket.removeListener('data', onData)
+            fastify.server.removeListener('request', onRequest)
+          }
+        }
+        fastify.server.on('request', onRequest)
+
+        socket.once('close', () => {
+          socket.removeListener('data', onData)
+          fastify.server.removeListener('request', onRequest)
+          if (hasRequest) {
+            return
+          }
+
+          const parsedHttp = parsePartialHttp(partialData)
+          const req = createPartialLogRequest(fastify, socket, parsedHttp, startTime)
+
+          doRequestLog(req, {
+            excludeUrls: options.excludeUrls,
+            statusCode: 'ABORTED CONN',
+            responseTime: (Date.now() - req.startTime) / 1000,
+          })
+        })
+      })
+
       fastify.addHook('onRequest', async (req, res) => {
         req.startTime = Date.now()
 
@@ -95,7 +152,7 @@ export const logRequest = (options: RequestLoggerOptions) =>
 interface LogRequestOptions {
   reply?: FastifyReply
   excludeUrls?: string[]
-  statusCode: number | 'ABORTED REQ' | 'ABORTED RES'
+  statusCode: number | 'ABORTED REQ' | 'ABORTED RES' | 'ABORTED CONN'
   responseTime: number
 }
 
@@ -142,3 +199,34 @@ function getFirstDefined<T>(...values: any[]): T | undefined {
   }
   return undefined
 }
+
+/**
+ * Creates a minimal FastifyRequest from partial HTTP data.
+ * Used for consistent logging when request parsing fails.
+ */
+export function createPartialLogRequest(
+  fastify: FastifyInstance,
+  socket: Socket,
+  httpData: PartialHttpData,
+  startTime: number
+) {
+  return {
+    method: httpData.method,
+    url: httpData.url,
+    headers: httpData.headers,
+    ip: socket.remoteAddress || 'unknown',
+    id: 'no-request',
+    log: fastify.log.child({
+      tenantId: httpData.tenantId,
+      project: httpData.tenantId,
+      reqId: 'no-request',
+      appVersion: version,
+      dataLength: httpData.length,
+    }),
+    startTime,
+    tenantId: httpData.tenantId,
+    raw: {},
+    routeOptions: { config: {} },
+    resources: [],
+  } as unknown as FastifyRequest
+}

src/internal/http/index.ts

@@ -1 +1,2 @@
 export * from './agent'
+export * from './partial-http-parser'

src/internal/http/partial-http-parser.ts

@@ -0,0 +1,80 @@
+import { getConfig } from '../../config'
+
+const { isMultitenant, requestXForwardedHostRegExp } = getConfig()
+
+const REQUEST_LINE_REGEX = /^([A-Z]+)\s+(\S+)(?:\s+HTTP\/[\d.]+)?$/i
+const LINE_SPLIT_REGEX = /\r?\n/
+// Validate header name (RFC 7230 token characters)
+const HEADER_NAME_REGEX = /^[a-z0-9!#$%&'*+\-.^_`|~]+$/
+
+const MAX_HEADER_LINES = 100
+
+export interface PartialHttpData {
+  method: string
+  url: string
+  headers: Record<string, string>
+  tenantId: string
+  length: number
+}
+
+/**
+ * Parses partial HTTP request data from raw buffers.
+ * Returns defaults if parsing fails.
+ */
+export function parsePartialHttp(dataChunks: Buffer[]): PartialHttpData {
+  const result: PartialHttpData = {
+    method: 'UNKNOWN',
+    url: '/',
+    headers: {},
+    tenantId: isMultitenant ? 'unknown' : 'storage-single-tenant',
+    length: 0,
+  }
+
+  if (dataChunks.length === 0) {
+    return result
+  }
+
+  try {
+    const partialData = Buffer.concat(dataChunks).toString('utf8')
+    const lines = partialData.split(LINE_SPLIT_REGEX)
+    result.length = partialData.length
+
+    // Parse request line: "METHOD /path HTTP/version"
+    if (lines[0]) {
+      const requestLine = lines[0].match(REQUEST_LINE_REGEX)
+      if (requestLine) {
+        result.method = requestLine[1].toUpperCase()
+        result.url = requestLine[2]
+      }
+    }
+
+    // Parse headers (skip line 0, limit total lines)
+    const headerLineLimit = Math.min(lines.length, MAX_HEADER_LINES + 1)
+    for (let i = 1; i < headerLineLimit; i++) {
+      const line = lines[i]
+      if (!line || line.trim() === '') continue
+
+      const colonIndex = line.indexOf(':')
+      if (colonIndex > 0) {
+        const field = line.substring(0, colonIndex).trim().toLowerCase()
+        const value = line.substring(colonIndex + 1).trim()
+        if (HEADER_NAME_REGEX.test(field)) {
+          result.headers[field] = value
+        }
+      }
+    }
+
+    // Extract tenantId from x-forwarded-host if multitenant
+    if (isMultitenant && requestXForwardedHostRegExp && result.headers['x-forwarded-host']) {
+      const match = result.headers['x-forwarded-host'].match(requestXForwardedHostRegExp)
+      if (match && match[1]) {
+        result.tenantId = match[1]
+      }
+    }
+  } catch {
+    // Parsing failed - return defaults
+    // This catches malformed UTF-8, regex errors, etc.
+  }
+
+  return result
+}