24025892439

Committed 06 Apr 2026 09:04AM UTC coverage: 80.536% (+0.02%) from 80.515%

Build # 24025892439

Build Type

push

github

Committed by

web-flow

Commit Message

fix: report errors in orphan and test entity expansion (#965)

Coverage Stats

3145 of 4079 branches covered (77.1%)

Branch coverage included in aggregate %.

135 of 188 new or added lines in 5 files covered. (71.81%)

3 existing lines in 1 file now uncovered.

30055 of 37145 relevant lines covered (80.91%)

311.64 hits per line

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

42.15

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

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

supabase / storage / 24025892439

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