28793654629

Committed 06 Jul 2026 01:05PM UTC coverage: 48.793% (-30.2%) from 79.041%

Build # 28793654629

Build Type

Pull #1206

github

Committed by

web-flow

Commit Message

Merge 056d0b5f9 into d54b900bf

Pull Request Pull Request #1206: fix: leak of request into pools

Coverage Stats

3210 of 7101 branches covered (45.2%)

Branch coverage included in aggregate %.

2 of 2 new or added lines in 2 files covered. (100.0%)

3858 existing lines in 164 files now uncovered.

6208 of 12201 relevant lines covered (50.88%)

63.5 hits per line

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

6.73

/src/http/routes/admin/objects.ts

import { render } from '@internal/errors'
import { logSchema } from '@internal/monitoring'
import { ObjectScanner } from '@storage/scanner/scanner'
import { FastifyInstance, RequestGenericInterface } from 'fastify'
import { FastifyReply } from 'fastify/types/reply'
import { dbSuperUser, storage } from '../../plugins'
import { registerApiKeyAuth } from '../../plugins/apikey'

const listOrphanedObjects = {
  description: 'List Orphaned Objects',
  params: {
    type: 'object',
    properties: {
      tenantId: { type: 'string' },
      bucketId: { type: 'string' },
    },
    required: ['tenantId', 'bucketId'],
  },
  query: {
    type: 'object',
    properties: {
      before: { type: 'string' },
      keepTmpTable: { type: 'boolean' },
    },
  },
} as const

const syncOrphanedObjects = {
  description: 'Sync Orphaned Objects',
  params: {
    type: 'object',
    properties: {
      tenantId: { type: 'string' },
      bucketId: { type: 'string' },
    },
    required: ['tenantId', 'bucketId'],
  },
  body: {
    type: 'object',
    properties: {
      deleteDbKeys: { type: 'boolean' },
      deleteS3Keys: { type: 'boolean' },
      tmpTable: { type: 'string' },
    },
  },
  optional: ['deleteDbKeys', 'deleteS3Keys'],
} as const

interface ListOrphanObjectsRequest extends RequestGenericInterface {
  Params: {
    tenantId: string
    bucketId: string
  }
  Querystring: {
    before?: string
    keepTmpTable?: boolean
  }
}

interface SyncOrphanObjectsRequest extends RequestGenericInterface {
  Params: {
    tenantId: string
    bucketId: string
  }
  Body: {
    deleteDbKeys?: boolean
    deleteS3Keys?: boolean
    before?: string
    tmpTable?: string
    keepTmpTable?: boolean
  }
}

export default async function routes(fastify: FastifyInstance) {
  registerApiKeyAuth(fastify)
  fastify.register(dbSuperUser, {
    disableHostCheck: true,
  })
  fastify.register(storage)

  fastify.get<ListOrphanObjectsRequest>(
    '/:tenantId/buckets/:bucketId/orphan-objects',
    {
      schema: { ...listOrphanedObjects, tags: ['object'] },
    },
    async (req, reply) => {
      const bucket = req.params.bucketId
      let before = req.query.before ? new Date(req.query.before as string) : undefined

      if (before && isNaN(before.getTime())) {
        return reply.status(400).send({
          error: 'Invalid date format',
        })
      }
      if (!before) {
        before = new Date()
        before.setHours(before.getHours() - 1)
      }

      const scanner = new ObjectScanner(req.storage)
      const orphanObjects = scanner.listOrphaned(bucket, {
        signal: req.signals.disconnect.signal,
        before,
        keepTmpTable: Boolean(req.query.keepTmpTable),
      })

      reply.header('Content-Type', 'application/x-ndjson; charset=utf-8')

      // Do not let the connection time out, periodically send
      // a ping message to keep the connection alive
      const respPing = ping(reply)

      try {
        for await (const result of orphanObjects) {
          if (result.value.length > 0) {
            respPing.update()
            writeNdjson(reply, {
              ...result,
              event: 'data',
            })
          }
        }
      } catch (e) {
        logSchema.error(req.log, 'list orphaned objects stream failed', {
          type: 'orphan',
          tenantId: req.tenantId,
          project: req.tenantId,
          reqId: req.id,
          sbReqId: req.sbReqId,
          error: e,
          metadata: JSON.stringify({ bucket }),
        })
        writeNdjson(reply, {
          event: 'error',
          error: render(e),
        })
        return
      } finally {
        respPing.clear()
        endNdjson(reply)
      }
    }
  )

  fastify.delete<SyncOrphanObjectsRequest>(
    '/:tenantId/buckets/:bucketId/orphan-objects',
    {
      schema: { ...syncOrphanedObjects, tags: ['object'] },
    },
    async (req, reply) => {
      if (!req.body.deleteDbKeys && !req.body.deleteS3Keys) {
        return reply.status(400).send({
          error: 'At least one of deleteDbKeys or deleteS3Keys must be set to true',
        })
      }

      const bucket = `${req.params.bucketId}`
      let before = req.body.before ? new Date(req.body.before as string) : undefined

      if (before && isNaN(before.getTime())) {
        return reply.status(400).send({
          error: 'Invalid date format',
        })
      }
      if (!before) {
        before = new Date()
        before.setHours(before.getHours() - 1)
      }

      reply.header('Content-Type', 'application/x-ndjson; charset=utf-8')

      const respPing = ping(reply)

      try {
        const scanner = new ObjectScanner(req.storage)
        const result = scanner.deleteOrphans(bucket, {
          deleteDbKeys: req.body.deleteDbKeys,
          deleteS3Keys: req.body.deleteS3Keys,
          signal: req.signals.disconnect.signal,
          before,
          tmpTable: req.body.tmpTable,
        })

        for await (const deleted of result) {
          respPing.update()
          writeNdjson(reply, {
            ...deleted,
            event: 'data',
          })
        }
      } catch (e) {
        logSchema.error(req.log, 'delete orphaned objects stream failed', {
          type: 'orphan',
          tenantId: req.tenantId,
          project: req.tenantId,
          reqId: req.id,
          sbReqId: req.sbReqId,
          error: e,
          metadata: JSON.stringify({ bucket }),
        })
        writeNdjson(reply, {
          event: 'error',
          error: render(e),
        })
        return
      } finally {
        respPing.clear()
        endNdjson(reply)
      }
    }
  )
}

function canWriteNdjson(reply: FastifyReply) {
  return !reply.raw.destroyed && !reply.raw.writableEnded
}

function writeNdjson(reply: FastifyReply, payload: unknown) {
  if (!canWriteNdjson(reply)) {
    return false
  }

  try {
    reply.raw.write(JSON.stringify(payload) + '\n')
    return true
  } catch {
    return false
  }
}

function endNdjson(reply: FastifyReply) {
  if (!canWriteNdjson(reply)) {
    return
  }

  try {
    reply.raw.end()
  } catch {}
}

// Occasionally write a ping message to the response stream
function ping(reply: FastifyReply) {
  let lastSend = undefined as Date | undefined
  const clearPing = setInterval(() => {
    const fiveSecondsEarly = new Date()
    fiveSecondsEarly.setSeconds(fiveSecondsEarly.getSeconds() - 5)

    if (!lastSend || (lastSend && lastSend < fiveSecondsEarly)) {
      lastSend = new Date()
      writeNdjson(reply, {
        event: 'ping',
      })
    }
  }, 1000 * 10)

  return {
    clear: () => clearInterval(clearPing),
    update: () => {
      lastSend = new Date()
    },
  }
}

1	import { render } from '@internal/errors'
2	import { logSchema } from '@internal/monitoring'
3	import { ObjectScanner } from '@storage/scanner/scanner'
4	import { FastifyInstance, RequestGenericInterface } from 'fastify'
5	import { FastifyReply } from 'fastify/types/reply'
6	import { dbSuperUser, storage } from '../../plugins'
7	import { registerApiKeyAuth } from '../../plugins/apikey'
8
9	const listOrphanedObjects = {	2✔
10	description: 'List Orphaned Objects',
11	params: {
12	type: 'object',
13	properties: {
14	tenantId: { type: 'string' },
15	bucketId: { type: 'string' },
16	},
17	required: ['tenantId', 'bucketId'],
18	},
19	query: {
20	type: 'object',
21	properties: {
22	before: { type: 'string' },
23	keepTmpTable: { type: 'boolean' },
24	},
25	},
26	} as const
27
28	const syncOrphanedObjects = {	2✔
29	description: 'Sync Orphaned Objects',
30	params: {
31	type: 'object',
32	properties: {
33	tenantId: { type: 'string' },
34	bucketId: { type: 'string' },
35	},
36	required: ['tenantId', 'bucketId'],
37	},
38	body: {
39	type: 'object',
40	properties: {
41	deleteDbKeys: { type: 'boolean' },
42	deleteS3Keys: { type: 'boolean' },
43	tmpTable: { type: 'string' },
44	},
45	},
46	optional: ['deleteDbKeys', 'deleteS3Keys'],
47	} as const
48
49	interface ListOrphanObjectsRequest extends RequestGenericInterface {
50	Params: {
51	tenantId: string
52	bucketId: string
53	}
54	Querystring: {
55	before?: string
56	keepTmpTable?: boolean
57	}
58	}
59
60	interface SyncOrphanObjectsRequest extends RequestGenericInterface {
61	Params: {
62	tenantId: string
63	bucketId: string
64	}
65	Body: {
66	deleteDbKeys?: boolean
67	deleteS3Keys?: boolean
68	before?: string
69	tmpTable?: string
70	keepTmpTable?: boolean
71	}
72	}
73
74	export default async function routes(fastify: FastifyInstance) {
75	registerApiKeyAuth(fastify)	2✔
76	fastify.register(dbSuperUser, {	2✔
77	disableHostCheck: true,
78	})
79	fastify.register(storage)	2✔
80
81	fastify.get<ListOrphanObjectsRequest>(	2✔
82	'/:tenantId/buckets/:bucketId/orphan-objects',
83	{
84	schema: { ...listOrphanedObjects, tags: ['object'] },
85	},
86	async (req, reply) => {
UNCOV 87	const bucket = req.params.bucketId	×
UNCOV 88	let before = req.query.before ? new Date(req.query.before as string) : undefined	×
89
UNCOV 90	if (before && isNaN(before.getTime())) {	×
UNCOV 91	return reply.status(400).send({	×
92	error: 'Invalid date format',
93	})
94	}
95	if (!before) {	×
96	before = new Date()	×
97	before.setHours(before.getHours() - 1)	×
98	}
99
100	const scanner = new ObjectScanner(req.storage)	×
101	const orphanObjects = scanner.listOrphaned(bucket, {	×
102	signal: req.signals.disconnect.signal,
103	before,
104	keepTmpTable: Boolean(req.query.keepTmpTable),
105	})
106
107	reply.header('Content-Type', 'application/x-ndjson; charset=utf-8')	×
108
109	// Do not let the connection time out, periodically send
110	// a ping message to keep the connection alive
111	const respPing = ping(reply)	×
112
113	try {	×
114	for await (const result of orphanObjects) {	×
115	if (result.value.length > 0) {	×
116	respPing.update()	×
117	writeNdjson(reply, {	×
118	...result,
119	event: 'data',
120	})
121	}
122	}
123	} catch (e) {
124	logSchema.error(req.log, 'list orphaned objects stream failed', {	×
125	type: 'orphan',
126	tenantId: req.tenantId,
127	project: req.tenantId,
128	reqId: req.id,
129	sbReqId: req.sbReqId,
130	error: e,
131	metadata: JSON.stringify({ bucket }),
132	})
133	writeNdjson(reply, {	×
134	event: 'error',
135	error: render(e),
136	})
137	return	×
138	} finally {
139	respPing.clear()	×
140	endNdjson(reply)	×
141	}
142	}
143	)
144
145	fastify.delete<SyncOrphanObjectsRequest>(	2✔
146	'/:tenantId/buckets/:bucketId/orphan-objects',
147	{
148	schema: { ...syncOrphanedObjects, tags: ['object'] },
149	},
150	async (req, reply) => {
UNCOV 151	if (!req.body.deleteDbKeys && !req.body.deleteS3Keys) {	×
UNCOV 152	return reply.status(400).send({	×
153	error: 'At least one of deleteDbKeys or deleteS3Keys must be set to true',
154	})
155	}
156
UNCOV 157	const bucket = `${req.params.bucketId}`	×
UNCOV 158	let before = req.body.before ? new Date(req.body.before as string) : undefined	×
159
UNCOV 160	if (before && isNaN(before.getTime())) {	×
UNCOV 161	return reply.status(400).send({	×
162	error: 'Invalid date format',
163	})
164	}
165	if (!before) {	×
166	before = new Date()	×
167	before.setHours(before.getHours() - 1)	×
168	}
169
170	reply.header('Content-Type', 'application/x-ndjson; charset=utf-8')	×
171
172	const respPing = ping(reply)	×
173
174	try {	×
175	const scanner = new ObjectScanner(req.storage)	×
176	const result = scanner.deleteOrphans(bucket, {	×
177	deleteDbKeys: req.body.deleteDbKeys,
178	deleteS3Keys: req.body.deleteS3Keys,
179	signal: req.signals.disconnect.signal,
180	before,
181	tmpTable: req.body.tmpTable,
182	})
183
184	for await (const deleted of result) {	×
185	respPing.update()	×
186	writeNdjson(reply, {	×
187	...deleted,
188	event: 'data',
189	})
190	}
191	} catch (e) {
192	logSchema.error(req.log, 'delete orphaned objects stream failed', {	×
193	type: 'orphan',
194	tenantId: req.tenantId,
195	project: req.tenantId,
196	reqId: req.id,
197	sbReqId: req.sbReqId,
198	error: e,
199	metadata: JSON.stringify({ bucket }),
200	})
201	writeNdjson(reply, {	×
202	event: 'error',
203	error: render(e),
204	})
205	return	×
206	} finally {
207	respPing.clear()	×
208	endNdjson(reply)	×
209	}
210	}
211	)
212	}
213
214	function canWriteNdjson(reply: FastifyReply) {
215	return !reply.raw.destroyed && !reply.raw.writableEnded	×
216	}
217
218	function writeNdjson(reply: FastifyReply, payload: unknown) {
219	if (!canWriteNdjson(reply)) {	×
220	return false	×
221	}
222
223	try {	×
224	reply.raw.write(JSON.stringify(payload) + '\n')	×
225	return true	×
226	} catch {
227	return false	×
228	}
229	}
230
231	function endNdjson(reply: FastifyReply) {
232	if (!canWriteNdjson(reply)) {	×
233	return	×
234	}
235
236	try {	×
237	reply.raw.end()	×
238	} catch {}
239	}
240
241	// Occasionally write a ping message to the response stream
242	function ping(reply: FastifyReply) {
243	let lastSend = undefined as Date \| undefined	×
244	const clearPing = setInterval(() => {	×
245	const fiveSecondsEarly = new Date()	×
246	fiveSecondsEarly.setSeconds(fiveSecondsEarly.getSeconds() - 5)	×
247
248	if (!lastSend \|\| (lastSend && lastSend < fiveSecondsEarly)) {	×
249	lastSend = new Date()	×
250	writeNdjson(reply, {	×
251	event: 'ping',
252	})
253	}
254	}, 1000 * 10)
255
256	return {	×
257	clear: () => clearInterval(clearPing),	×
258	update: () => {
259	lastSend = new Date()	×
260	},
261	}
262	}

supabase / storage / 28793654629

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