23000572934

Committed 12 Mar 2026 11:50AM UTC coverage: 76.311% (+0.03%) from 76.282%

Build # 23000572934

Build Type

Pull #783

github

Committed by

web-flow

Commit Message

Merge 3b43514ee into 90936a809

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

Coverage Stats

4041 of 5749 branches covered (70.29%)

Branch coverage included in aggregate %.

89 of 102 new or added lines in 1 file covered. (87.25%)

6 existing lines in 1 file now uncovered.

26878 of 34768 relevant lines covered (77.31%)

185.42 hits per line

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

82.9

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

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[]
}

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) => {
      // 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) => {
        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
        })

        // Track when data arrives for a potential request
        const onData = () => {
          // Start tracking a new potential request when we receive data after a completed one
          if (!waitingForRequest) {
            waitingForRequest = true
            currentRequestStart = Date.now()
            pendingRequestLogged = false
          }
        }
        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 || pendingRequestLogged) {
            return
          }

          const req = createPartialLogRequest(fastify, socket, 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 && typeof req.opentelemetry === 'function') {
          req.opentelemetry()?.span?.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 {
          if (typeof req.opentelemetry === 'function') {
            req.opentelemetry()?.span?.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 for logging aborted connections.
 * Used when connection closes before a complete HTTP request is received.
 */
export function createPartialLogRequest(
  fastify: FastifyInstance,
  socket: Socket,
  startTime: number
) {
  return {
    method: 'UNKNOWN',
    headers: {},
    url: '/',
    ip: socket.remoteAddress || 'unknown',
    id: 'no-request',
    log: fastify.log.child({
      reqId: 'no-request',
      appVersion: version,
    }),
    startTime,
    raw: {},
    routeOptions: { config: {} },
    resources: [],
  } as unknown as FastifyRequest
}

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

supabase / storage / 23000572934

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