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

Author: itslenny

src/http/plugins/log-request.ts

@@ -1,7 +1,11 @@
+import { PartialHttpData, parsePartialHttp } from '@internal/http'
 import { logger, logSchema, redactQueryParamFromRequest } from '@internal/monitoring'
+import { FastifyInstance } from 'fastify'
 import { FastifyReply } from 'fastify/types/reply'
 import { FastifyRequest } from 'fastify/types/request'
 import fastifyPlugin from 'fastify-plugin'
+import { Socket } from 'net'
+import { getConfig } from '../../config'
 
 interface RequestLoggerOptions {
   excludeUrls?: string[]
@@ -23,13 +27,95 @@ declare module 'fastify' {
   }
 }
 
+const { version } = getConfig()
+
 /**
  * Request logger plugin
  * @param options
  */
 export const logRequest = (options: RequestLoggerOptions) =>
   fastifyPlugin(
     async (fastify) => {
+      // Used to track cleanup functions per socket
+      const socketCleanupMap = new WeakMap<Socket, () => void>()
+      const cleanupSocketListeners = (socket: Socket) => {
+        const cleanup = socketCleanupMap.get(socket)
+        if (cleanup) {
+          socketCleanupMap.delete(socket)
+          cleanup()
+        }
+      }
+
+      // Watch for connections that timeout or disconnect before complete HTTP headers are received
+      // For keep-alive connections, track each potential request independently
+      const onConnection = (socket: Socket) => {
+        const captureByteLimit = 2048
+        let currentRequestData: Buffer[] = []
+        let currentRequestDataSize = 0
+        let currentRequestStart = Date.now()
+        let waitingForRequest = false
+        let pendingRequestLogged = false
+
+        // Store cleanup function so hooks can access it
+        socketCleanupMap.set(socket, () => {
+          pendingRequestLogged = true
+          waitingForRequest = false
+          currentRequestData = []
+          currentRequestDataSize = 0
+        })
+
+        // Capture partial data sent before connection closes
+        const onData = (chunk: Buffer) => {
+          // Start tracking a new potential request when we receive data after a completed one
+          if (!waitingForRequest) {
+            waitingForRequest = true
+            currentRequestData = []
+            currentRequestDataSize = 0
+            currentRequestStart = Date.now()
+            pendingRequestLogged = false
+          }
+
+          const remaining = captureByteLimit - currentRequestDataSize
+          if (remaining > 0) {
+            const slicedChunk = chunk.subarray(0, Math.min(chunk.length, remaining))
+            currentRequestData.push(slicedChunk)
+            currentRequestDataSize += slicedChunk.length
+          }
+        }
+        socket.on('data', onData)
+
+        // Remove data listener on socket error to prevent listener leak
+        socket.once('error', () => {
+          socket.removeListener('data', onData)
+        })
+
+        socket.once('close', () => {
+          socket.removeListener('data', onData)
+          socketCleanupMap.delete(socket)
+
+          // Only log if we were waiting for a request that was never properly logged
+          if (!waitingForRequest || currentRequestData.length === 0 || pendingRequestLogged) {
+            return
+          }
+
+          const parsedHttp = parsePartialHttp(currentRequestData)
+          const req = createPartialLogRequest(fastify, socket, parsedHttp, currentRequestStart)
+
+          doRequestLog(req, {
+            excludeUrls: options.excludeUrls,
+            statusCode: 'ABORTED CONN',
+            responseTime: (Date.now() - req.startTime) / 1000,
+          })
+        })
+      }
+
+      fastify.server.on('connection', onConnection)
+
+      // Clean up on close
+      fastify.addHook('onClose', async () => {
+        fastify.server.removeListener('connection', onConnection)
+      })
+
       fastify.addHook('onRequest', async (req, res) => {
         req.startTime = Date.now()
 
@@ -40,6 +126,7 @@ export const logRequest = (options: RequestLoggerOptions) =>
               statusCode: 'ABORTED REQ',
               responseTime: (Date.now() - req.startTime) / 1000,
             })
+            cleanupSocketListeners(req.raw.socket)
             return
           }
 
@@ -49,6 +136,7 @@ export const logRequest = (options: RequestLoggerOptions) =>
               statusCode: 'ABORTED RES',
               responseTime: (Date.now() - req.startTime) / 1000,
             })
+            cleanupSocketListeners(req.raw.socket)
           }
         })
       })
@@ -94,6 +182,9 @@ export const logRequest = (options: RequestLoggerOptions) =>
           responseTime: reply.elapsedTime,
           executionTime: req.executionTime,
         })
+
+        // Mark request as logged so socket close handler doesn't log it again
+        cleanupSocketListeners(req.raw.socket)
       })
     },
     { name: 'log-request' }
@@ -102,7 +193,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
   executionTime?: number
 }
@@ -179,3 +270,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
+}