24848158243

Committed 23 Apr 2026 05:03PM UTC coverage: 71.255% (+1.3%) from 69.952%

Build # 24848158243

Build Type

push

github

Committed by

web-flow

Commit Message

feat: add sb-request-id logging (#1041)

* feat: add sb-request-id logging

Also propagate the existing `reqId` where missed.

Signed-off-by: ferhat elmas <elmas.ferhat@gmail.com>

* chore: system tenant

Signed-off-by: ferhat elmas <elmas.ferhat@gmail.com>

---------

Signed-off-by: ferhat elmas <elmas.ferhat@gmail.com>

Coverage Stats

3480 of 5443 branches covered (63.94%)

Branch coverage included in aggregate %.

81 of 89 new or added lines in 25 files covered. (91.01%)

6 existing lines in 4 files now uncovered.

7333 of 9732 relevant lines covered (75.35%)

377.75 hits per line

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

28.85

/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 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) {
        logSchema.error(req.log, 'list orphaned objects stream failed', {
          type: 'orphan',
          error: e,
          project: req.params.tenantId,
          metadata: JSON.stringify({ bucket }),
          sbReqId: req.sbReqId,
        })
        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',
          error: e,
          project: req.params.tenantId,
          metadata: JSON.stringify({ bucket }),
          sbReqId: req.sbReqId,
        })
        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 apiKey from '../../plugins/apikey'
8
9	const listOrphanedObjects = {	25✔
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 = {	25✔
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	fastify.register(apiKey)	11✔
76	fastify.register(dbSuperUser, {	11✔
77	disableHostCheck: true,
78	maxConnections: 5,
79	})
80	fastify.register(storage)	11✔
81
82	fastify.get<ListOrphanObjectsRequest>(	11✔
83	'/:tenantId/buckets/:bucketId/orphan-objects',
84	{
85	schema: { ...listOrphanedObjects, tags: ['object'] },
86	},
87	async (req, reply) => {
88	const bucket = req.params.bucketId	1✔
89	let before = req.query.before ? new Date(req.query.before as string) : undefined	1!
90
91	if (before && isNaN(before.getTime())) {	1!
92	return reply.status(400).send({	1✔
93	error: 'Invalid date format',
94	})
95	}
96	if (!before) {	×
97	before = new Date()	×
98	before.setHours(before.getHours() - 1)	×
99	}
100
101	const scanner = new ObjectScanner(req.storage)	×
102	const orphanObjects = scanner.listOrphaned(bucket, {	×
103	signal: req.signals.disconnect.signal,
104	before,
105	keepTmpTable: Boolean(req.query.keepTmpTable),
106	})
107
108	reply.header('Content-Type', 'application/x-ndjson; charset=utf-8')	×
109
110	// Do not let the connection time out, periodically send
111	// a ping message to keep the connection alive
112	const respPing = ping(reply)	×
113
114	try {	×
115	for await (const result of orphanObjects) {	×
116	if (result.value.length > 0) {	×
117	respPing.update()	×
118	writeNdjson(reply, {	×
119	...result,
120	event: 'data',
121	})
122	}
123	}
124	} catch (e) {
NEW 125	logSchema.error(req.log, 'list orphaned objects stream failed', {	×
126	type: 'orphan',
127	error: e,
128	project: req.params.tenantId,
129	metadata: JSON.stringify({ bucket }),
130	sbReqId: req.sbReqId,
131	})
UNCOV 132	writeNdjson(reply, {	×
133	event: 'error',
134	error: render(e),
135	})
136	return	×
137	} finally {
138	respPing.clear()	×
139	endNdjson(reply)	×
140	}
141	}
142	)
143
144	fastify.delete<SyncOrphanObjectsRequest>(	11✔
145	'/:tenantId/buckets/:bucketId/orphan-objects',
146	{
147	schema: { ...syncOrphanedObjects, tags: ['object'] },
148	},
149	async (req, reply) => {
150	if (!req.body.deleteDbKeys && !req.body.deleteS3Keys) {	2✔
151	return reply.status(400).send({	1✔
152	error: 'At least one of deleteDbKeys or deleteS3Keys must be set to true',
153	})
154	}
155
156	const bucket = `${req.params.bucketId}`	1✔
157	let before = req.body.before ? new Date(req.body.before as string) : undefined	1!
158
159	if (before && isNaN(before.getTime())) {	2✔
160	return reply.status(400).send({	1✔
161	error: 'Invalid date format',
162	})
163	}
164	if (!before) {	×
165	before = new Date()	×
166	before.setHours(before.getHours() - 1)	×
167	}
168
169	reply.header('Content-Type', 'application/x-ndjson; charset=utf-8')	×
170
171	const respPing = ping(reply)	×
172
173	try {	×
174	const scanner = new ObjectScanner(req.storage)	×
175	const result = scanner.deleteOrphans(bucket, {	×
176	deleteDbKeys: req.body.deleteDbKeys,
177	deleteS3Keys: req.body.deleteS3Keys,
178	signal: req.signals.disconnect.signal,
179	before,
180	tmpTable: req.body.tmpTable,
181	})
182
183	for await (const deleted of result) {	×
184	respPing.update()	×
185	writeNdjson(reply, {	×
186	...deleted,
187	event: 'data',
188	})
189	}
190	} catch (e) {
NEW 191	logSchema.error(req.log, 'delete orphaned objects stream failed', {	×
192	type: 'orphan',
193	error: e,
194	project: req.params.tenantId,
195	metadata: JSON.stringify({ bucket }),
196	sbReqId: req.sbReqId,
197	})
UNCOV 198	writeNdjson(reply, {	×
199	event: 'error',
200	error: render(e),
201	})
202	return	×
203	} finally {
204	respPing.clear()	×
205	endNdjson(reply)	×
206	}
207	}
208	)
209	}
210
211	function canWriteNdjson(reply: FastifyReply) {
212	return !reply.raw.destroyed && !reply.raw.writableEnded	×
213	}
214
215	function writeNdjson(reply: FastifyReply, payload: unknown) {
216	if (!canWriteNdjson(reply)) {	×
217	return false	×
218	}
219
220	try {	×
221	reply.raw.write(JSON.stringify(payload) + '\n')	×
222	return true	×
223	} catch {
224	return false	×
225	}
226	}
227
228	function endNdjson(reply: FastifyReply) {
229	if (!canWriteNdjson(reply)) {	×
230	return	×
231	}
232
233	try {	×
234	reply.raw.end()	×
235	} catch {}
236	}
237
238	// Occasionally write a ping message to the response stream
239	function ping(reply: FastifyReply) {
240	let lastSend = undefined as Date \| undefined	×
241	const clearPing = setInterval(() => {	×
242	const fiveSecondsEarly = new Date()	×
243	fiveSecondsEarly.setSeconds(fiveSecondsEarly.getSeconds() - 5)	×
244
245	if (!lastSend \|\| (lastSend && lastSend < fiveSecondsEarly)) {	×
246	lastSend = new Date()	×
247	writeNdjson(reply, {	×
248	event: 'ping',
249	})
250	}
251	}, 1000 * 10)
252
253	return {	×
254	clear: () => clearInterval(clearPing),	×
255	update: () => {
256	lastSend = new Date()	×
257	},
258	}
259	}

supabase / storage / 24848158243

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