22756907554

Committed 06 Mar 2026 09:14AM UTC coverage: 76.118% (+0.04%) from 76.078%

Build # 22756907554

Build Type

Pull #783

github

Committed by

web-flow

Commit Message

Merge 14342a733 into bb602aa99

Pull Request Pull Request #783: fix: log connections that timeout, abort, or send unparsable data

Coverage Stats

3963 of 5677 branches covered (69.81%)

Branch coverage included in aggregate %.

181 of 205 new or added lines in 3 files covered. (88.29%)

3 existing lines in 1 file now uncovered.

26807 of 34747 relevant lines covered (77.15%)

187.33 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

84.74

/src/http/plugins/log-request.ts

import { PartialHttpData, parsePartialHttp } from '@internal/http'
import { logger, logSchema, redactQueryParamFromRequest } from '@internal/monitoring'
import { trace } from '@opentelemetry/api'
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[]
}

declare module 'fastify' {
  interface FastifyRequest {
    executionError?: Error
    operation?: { type: string }
    resources?: string[]
    startTime: number
    executionTime?: number
  }

  interface FastifyContextConfig {
    operation?: { type: string }
    resources?: (req: FastifyRequest<any>) => string[]
    logMetadata?: (req: FastifyRequest<any>) => Record<string, unknown>
  }
}

const { version } = getConfig()

/**
 * Request logger plugin
 * @param options
 */
export const logRequest = (options: RequestLoggerOptions) =>
  fastifyPlugin(
    async (fastify) => {
      // WeakMap to track cleanup functions per socket without modifying socket object
      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 in WeakMap 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()

        res.raw.once('close', () => {
          if (req.raw.aborted) {
            doRequestLog(req, {
              excludeUrls: options.excludeUrls,
              statusCode: 'ABORTED REQ',
              responseTime: (Date.now() - req.startTime) / 1000,
            })
            cleanupSocketListeners(req.raw.socket)
            return
          }

          if (!res.raw.writableFinished) {
            doRequestLog(req, {
              excludeUrls: options.excludeUrls,
              statusCode: 'ABORTED RES',
              responseTime: (Date.now() - req.startTime) / 1000,
            })
            cleanupSocketListeners(req.raw.socket)
          }
        })
      })

      /**
       * Adds req.resources and req.operation to the request object
       */
      fastify.addHook('preHandler', async (req) => {
        const resourceFromParams = Object.values(req.params || {}).join('/')
        const resources = getFirstDefined<string[]>(
          req.resources,
          req.routeOptions.config.resources?.(req),
          (req.raw as any).resources,
          resourceFromParams ? [resourceFromParams] : ([] as string[])
        )

        if (resources && resources.length > 0) {
          resources.map((resource, index) => {
            if (!resource.startsWith('/')) {
              resources[index] = `/${resource}`
            }
          })
        }

        req.resources = resources
        req.operation = req.routeOptions.config.operation

        if (req.operation) {
          trace.getActiveSpan()?.setAttribute('http.operation', req.operation.type)
        }
      })

      fastify.addHook('onSend', async (req, _, payload) => {
        req.executionTime = Date.now() - req.startTime
        return payload
      })

      fastify.addHook('onResponse', async (req, reply) => {
        doRequestLog(req, {
          reply,
          excludeUrls: options.excludeUrls,
          statusCode: reply.statusCode,
          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' }
  )

interface LogRequestOptions {
  reply?: FastifyReply
  excludeUrls?: string[]
  statusCode: number | 'ABORTED REQ' | 'ABORTED RES' | 'ABORTED CONN'
  responseTime: number
  executionTime?: number
}

function doRequestLog(req: FastifyRequest, options: LogRequestOptions) {
  if (options.excludeUrls?.includes(req.url)) {
    return
  }

  const rMeth = req.method
  const rUrl = redactQueryParamFromRequest(req, [
    'token',
    'X-Amz-Credential',
    'X-Amz-Signature',
    'X-Amz-Security-Token',
  ])
  const uAgent = req.headers['user-agent']
  const rId = req.id
  const cIP = req.ip
  const statusCode = options.statusCode
  const error = (req.raw as any).executionError || req.executionError
  const tenantId = req.tenantId

  let reqMetadata: Record<string, unknown> = {}

  if (req.routeOptions.config.logMetadata) {
    try {
      reqMetadata = req.routeOptions.config.logMetadata(req)

      if (reqMetadata) {
        try {
          trace.getActiveSpan()?.setAttribute('http.metadata', JSON.stringify(reqMetadata))
        } catch (e) {
          // do nothing
          logSchema.warning(logger, 'Failed to serialize log metadata', {
            type: 'otel',
            error: e,
          })
        }
      }
    } catch (e) {
      logSchema.error(logger, 'Failed to get log metadata', {
        type: 'request',
        error: e,
      })
    }
  }

  const buildLogMessage = `${tenantId} | ${rMeth} | ${statusCode} | ${cIP} | ${rId} | ${rUrl} | ${uAgent}`

  logSchema.request(req.log, buildLogMessage, {
    type: 'request',
    req,
    reqMetadata,
    res: options.reply,
    responseTime: options.responseTime,
    executionTime: options.executionTime,
    error,
    owner: req.owner,
    role: req.jwtPayload?.role,
    resources: req.resources,
    operation: req.operation?.type ?? req.routeOptions.config.operation?.type,
    serverTimes: req.serverTimings,
  })
}

function getFirstDefined<T>(...values: any[]): T | undefined {
  for (const value of values) {
    if (value !== undefined) {
      return value
    }
  }
  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
}

1	import { PartialHttpData, parsePartialHttp } from '@internal/http'	2✔
2	import { logger, logSchema, redactQueryParamFromRequest } from '@internal/monitoring'	2✔
3	import { trace } from '@opentelemetry/api'	2✔
4	import { FastifyInstance } from 'fastify'	2✔
5	import { FastifyReply } from 'fastify/types/reply'	2✔
6	import { FastifyRequest } from 'fastify/types/request'	2✔
7	import fastifyPlugin from 'fastify-plugin'	2✔
8	import { Socket } from 'net'	2✔
9	import { getConfig } from '../../config'	2✔
10		2✔
11	interface RequestLoggerOptions {	2✔
12	excludeUrls?: string[]	2✔
13	}	2✔
14		2✔
15	declare module 'fastify' {	2✔
16	interface FastifyRequest {	2✔
17	executionError?: Error	2✔
18	operation?: { type: string }	2✔
19	resources?: string[]	2✔
20	startTime: number	2✔
21	executionTime?: number	2✔
22	}	2✔
23		2✔
24	interface FastifyContextConfig {	2✔
25	operation?: { type: string }	2✔
26	resources?: (req: FastifyRequest<any>) => string[]	2✔
27	logMetadata?: (req: FastifyRequest<any>) => Record<string, unknown>	2✔
28	}	2✔
29	}	2✔
30		2✔
31	const { version } = getConfig()	2✔
32		2✔
33	/**	2✔
34	* Request logger plugin	2✔
35	* @param options	2✔
36	*/	2✔
37	export const logRequest = (options: RequestLoggerOptions) =>	2✔
38	fastifyPlugin(	264✔
39	async (fastify) => {	264✔
40	// WeakMap to track cleanup functions per socket without modifying socket object	262✔
41	const socketCleanupMap = new WeakMap<Socket, () => void>()	262✔
42		262✔
43	const cleanupSocketListeners = (socket: Socket) => {	262✔
44	const cleanup = socketCleanupMap.get(socket)	564✔
45	if (cleanup) {	564!
NEW 46	socketCleanupMap.delete(socket)	×
NEW 47	cleanup()	×
NEW 48	}	×
49	}	564✔
50		262✔
51	// Watch for connections that timeout or disconnect before complete HTTP headers are received	262✔
52	// For keep-alive connections, track each potential request independently	262✔
53	const onConnection = (socket: Socket) => {	262✔
54	const captureByteLimit = 2048	4✔
55	let currentRequestData: Buffer[] = []	4✔
56	let currentRequestDataSize = 0	4✔
57	let currentRequestStart = Date.now()	4✔
58	let waitingForRequest = false	4✔
59	let pendingRequestLogged = false	4✔
60		4✔
61	// Store cleanup function in WeakMap so hooks can access it	4✔
62	socketCleanupMap.set(socket, () => {	4✔
63	pendingRequestLogged = true	4✔
64	waitingForRequest = false	4✔
65	currentRequestData = []	4✔
66	currentRequestDataSize = 0	4✔
67	})	4✔
68		4✔
69	// Capture partial data sent before connection closes	4✔
70	const onData = (chunk: Buffer) => {	4✔
71	// Start tracking a new potential request when we receive data after a completed one	4✔
72	if (!waitingForRequest) {	4✔
73	waitingForRequest = true	4✔
74	currentRequestData = []	4✔
75	currentRequestDataSize = 0	4✔
76	currentRequestStart = Date.now()	4✔
77	pendingRequestLogged = false	4✔
78	}	4✔
79		4✔
80	const remaining = captureByteLimit - currentRequestDataSize	4✔
81	if (remaining > 0) {	4✔
82	const slicedChunk = chunk.subarray(0, Math.min(chunk.length, remaining))	4✔
83	currentRequestData.push(slicedChunk)	4✔
84	currentRequestDataSize += slicedChunk.length	4✔
85	}	4✔
86	}	4✔
87	socket.on('data', onData)	4✔
88		4✔
89	// Remove data listener on socket error to prevent listener leak	4✔
90	socket.once('error', () => {	4✔
NEW 91	socket.removeListener('data', onData)	×
92	})	4✔
93		4✔
94	socket.once('close', () => {	4✔
95	socket.removeListener('data', onData)	4✔
96	socketCleanupMap.delete(socket)	4✔
97		4✔
98	// Only log if we were waiting for a request that was never properly logged	4✔
99	if (!waitingForRequest \|\| currentRequestData.length === 0 \|\| pendingRequestLogged) {	4!
100	return	4✔
101	}	4✔
NEW 102		×
NEW 103	const parsedHttp = parsePartialHttp(currentRequestData)	×
NEW 104	const req = createPartialLogRequest(fastify, socket, parsedHttp, currentRequestStart)	×
NEW 105		×
NEW 106	doRequestLog(req, {	×
NEW 107	excludeUrls: options.excludeUrls,	×
NEW 108	statusCode: 'ABORTED CONN',	×
NEW 109	responseTime: (Date.now() - req.startTime) / 1000,	×
NEW 110	})	×
111	})	4✔
112	}	4✔
113		262✔
114	fastify.server.on('connection', onConnection)	262✔
115		262✔
116	// Clean up on close	262✔
117	fastify.addHook('onClose', async () => {	262✔
118	fastify.server.removeListener('connection', onConnection)	262✔
119	})	262✔
120		262✔
121	fastify.addHook('onRequest', async (req, res) => {	262✔
122	req.startTime = Date.now()	282✔
123		282✔
124	res.raw.once('close', () => {	282✔
125	if (req.raw.aborted) {	282!
126	doRequestLog(req, {	×
127	excludeUrls: options.excludeUrls,	×
128	statusCode: 'ABORTED REQ',	×
129	responseTime: (Date.now() - req.startTime) / 1000,	×
130	})	×
NEW 131	cleanupSocketListeners(req.raw.socket)	×
132	return	×
133	}	×
134		282✔
135	if (!res.raw.writableFinished) {	282✔
136	doRequestLog(req, {	282✔
137	excludeUrls: options.excludeUrls,	282✔
138	statusCode: 'ABORTED RES',	282✔
139	responseTime: (Date.now() - req.startTime) / 1000,	282✔
140	})	282✔
141	cleanupSocketListeners(req.raw.socket)	282✔
142	}	282✔
143	})	282✔
144	})	262✔
145		262✔
146	/**	262✔
147	* Adds req.resources and req.operation to the request object	262✔
148	*/	262✔
149	fastify.addHook('preHandler', async (req) => {	262✔
150	const resourceFromParams = Object.values(req.params \|\| {}).join('/')	252!
151	const resources = getFirstDefined<string[]>(	252✔
152	req.resources,	252✔
153	req.routeOptions.config.resources?.(req),	252✔
154	(req.raw as any).resources,	252✔
155	resourceFromParams ? [resourceFromParams] : ([] as string[])	252✔
156	)	252✔
157		252✔
158	if (resources && resources.length > 0) {	252✔
159	resources.map((resource, index) => {	252✔
160	if (!resource.startsWith('/')) {	80,294✔
161	resources[index] = `/${resource}`	80,294✔
162	}	80,294✔
163	})	252✔
164	}	252✔
165		252✔
166	req.resources = resources	252✔
167	req.operation = req.routeOptions.config.operation	252✔
168		252✔
169	if (req.operation) {	252✔
170	trace.getActiveSpan()?.setAttribute('http.operation', req.operation.type)	252!
171	}	252✔
172	})	262✔
173		262✔
174	fastify.addHook('onSend', async (req, _, payload) => {	262✔
175	req.executionTime = Date.now() - req.startTime	282✔
176	return payload	282✔
177	})	262✔
178		262✔
179	fastify.addHook('onResponse', async (req, reply) => {	262✔
180	doRequestLog(req, {	282✔
181	reply,	282✔
182	excludeUrls: options.excludeUrls,	282✔
183	statusCode: reply.statusCode,	282✔
184	responseTime: reply.elapsedTime,	282✔
185	executionTime: req.executionTime,	282✔
186	})	282✔
187		282✔
188	// Mark request as logged so socket close handler doesn't log it again	282✔
189	cleanupSocketListeners(req.raw.socket)	282✔
190	})	262✔
191	},	262✔
192	{ name: 'log-request' }	264✔
193	)	2✔
194		2✔
195	interface LogRequestOptions {	2✔
196	reply?: FastifyReply	2✔
197	excludeUrls?: string[]	2✔
198	statusCode: number \| 'ABORTED REQ' \| 'ABORTED RES' \| 'ABORTED CONN'	2✔
199	responseTime: number	2✔
200	executionTime?: number	2✔
201	}	2✔
202		2✔
203	function doRequestLog(req: FastifyRequest, options: LogRequestOptions) {	564✔
204	if (options.excludeUrls?.includes(req.url)) {	564!
205	return	×
206	}	×
207		564✔
208	const rMeth = req.method	564✔
209	const rUrl = redactQueryParamFromRequest(req, [	564✔
210	'token',	564✔
211	'X-Amz-Credential',	564✔
212	'X-Amz-Signature',	564✔
213	'X-Amz-Security-Token',	564✔
214	])	564✔
215	const uAgent = req.headers['user-agent']	564✔
216	const rId = req.id	564✔
217	const cIP = req.ip	564✔
218	const statusCode = options.statusCode	564✔
219	const error = (req.raw as any).executionError \|\| req.executionError	564✔
220	const tenantId = req.tenantId	564✔
221		564✔
222	let reqMetadata: Record<string, unknown> = {}	564✔
223		564✔
224	if (req.routeOptions.config.logMetadata) {	564✔
225	try {	44✔
226	reqMetadata = req.routeOptions.config.logMetadata(req)	44✔
227		44✔
228	if (reqMetadata) {	44✔
229	try {	44✔
230	trace.getActiveSpan()?.setAttribute('http.metadata', JSON.stringify(reqMetadata))	44!
231	} catch (e) {	44!
232	// do nothing	×
233	logSchema.warning(logger, 'Failed to serialize log metadata', {	×
234	type: 'otel',	×
235	error: e,	×
236	})	×
237	}	×
238	}	44✔
239	} catch (e) {	44!
240	logSchema.error(logger, 'Failed to get log metadata', {	×
241	type: 'request',	×
242	error: e,	×
243	})	×
244	}	×
245	}	44✔
246		564✔
247	const buildLogMessage = `${tenantId} \| ${rMeth} \| ${statusCode} \| ${cIP} \| ${rId} \| ${rUrl} \| ${uAgent}`	564✔
248		564✔
249	logSchema.request(req.log, buildLogMessage, {	564✔
250	type: 'request',	564✔
251	req,	564✔
252	reqMetadata,	564✔
253	res: options.reply,	564✔
254	responseTime: options.responseTime,	564✔
255	executionTime: options.executionTime,	564✔
256	error,	564✔
257	owner: req.owner,	564✔
258	role: req.jwtPayload?.role,	564✔
259	resources: req.resources,	564✔
260	operation: req.operation?.type ?? req.routeOptions.config.operation?.type,	564✔
261	serverTimes: req.serverTimings,	564✔
262	})	564✔
263	}	564✔
264		2✔
265	function getFirstDefined<T>(...values: any[]): T \| undefined {	252✔
266	for (const value of values) {	252✔
267	if (value !== undefined) {	900✔
268	return value	252✔
269	}	252✔
270	}	900✔
271	return undefined	×
272	}	×
273		2✔
274	/**	2✔
275	* Creates a minimal FastifyRequest from partial HTTP data.	2✔
276	* Used for consistent logging when request parsing fails.	2✔
277	*/	2✔
278	export function createPartialLogRequest(	8✔
279	fastify: FastifyInstance,	8✔
280	socket: Socket,	8✔
281	httpData: PartialHttpData,	8✔
282	startTime: number	8✔
283	) {	8✔
284	return {	8✔
285	method: httpData.method,	8✔
286	url: httpData.url,	8✔
287	headers: httpData.headers,	8✔
288	ip: socket.remoteAddress \|\| 'unknown',	8!
289	id: 'no-request',	8✔
290	log: fastify.log.child({	8✔
291	tenantId: httpData.tenantId,	8✔
292	project: httpData.tenantId,	8✔
293	reqId: 'no-request',	8✔
294	appVersion: version,	8✔
295	dataLength: httpData.length,	8✔
296	}),	8✔
297	startTime,	8✔
298	tenantId: httpData.tenantId,	8✔
299	raw: {},	8✔
300	routeOptions: { config: {} },	8✔
301	resources: [],	8✔
302	} as unknown as FastifyRequest	8✔
303	}	8✔

supabase / storage / 22756907554

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous