Skip to content

Add consistent uploadId/batchId log prefix on the worker's call path #114

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
DaxServer merged 1 commit into from
Jul 5, 2026
Merged

Add consistent uploadId/batchId log prefix on the worker's call path #114

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 8 additions & 6 deletions backend/src/__tests__/uploadClient.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -731,7 +731,7 @@ describe('MediaWikiClient.applySdc', () => {
const apiRequestMock = mock(async () => ({}))
// biome-ignore lint/suspicious/noExplicitAny: overriding private methods for testing
;(client as any).apiRequest = apiRequestMock
await client.applySdc('Photo.jpg', [{ mainsnak: {} }], null, 'summary')
await client.applySdc('Photo.jpg', [{ mainsnak: {} }], null, 'summary', '[1/1]')
expect(apiRequestMock).toHaveBeenCalled()
})

Expand All @@ -741,7 +741,9 @@ describe('MediaWikiClient.applySdc', () => {
;(client as any).getCsrfToken = mock(async () => 'token+\\')
// biome-ignore lint/suspicious/noExplicitAny: overriding private methods for testing
;(client as any).apiRequest = mock(async () => ({ error: { info: 'SDC error' } }))
await expect(client.applySdc('Photo.jpg', null, null, 'summary')).rejects.toThrow('SDC error')
await expect(client.applySdc('Photo.jpg', null, null, 'summary', '[1/1]')).rejects.toThrow(
'SDC error',
)
})

it('retries with a fresh token when the wbeditentity request returns badtoken', async () => {
Expand All @@ -758,7 +760,7 @@ describe('MediaWikiClient.applySdc', () => {
return {}
})

await client.applySdc('Test.jpg', null, null, 'summary')
await client.applySdc('Test.jpg', null, null, 'summary', '[1/1]')

expect(apiCall).toBe(2)
})
Expand Down Expand Up @@ -870,7 +872,7 @@ describe('MediaWikiClient.nullEdit retry', () => {
if (editAttempts <= 2) throw new Error('transient network error')
return {}
})
await expect(client.nullEdit('Photo.jpg')).resolves.toBeUndefined()
await expect(client.nullEdit('Photo.jpg', '[1/1]')).resolves.toBeUndefined()
expect(editAttempts).toBe(3)
} finally {
globalThis.setTimeout = origSetTimeout
Expand All @@ -888,7 +890,7 @@ describe('MediaWikiClient.nullEdit retry', () => {
if (params.action === 'query') return makeQueryResponse()
throw new Error('persistent error')
})
await expect(client.nullEdit('Photo.jpg')).rejects.toThrow('persistent error')
await expect(client.nullEdit('Photo.jpg', '[1/1]')).rejects.toThrow('persistent error')
} finally {
globalThis.setTimeout = origSetTimeout
}
Expand All @@ -909,7 +911,7 @@ describe('MediaWikiClient.nullEdit retry', () => {
return {}
})

await client.nullEdit('Test.jpg')
await client.nullEdit('Test.jpg', '[1/1]')

expect(apiCall).toBe(2)
})
Expand Down
4 changes: 4 additions & 0 deletions backend/src/core/logger.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,10 @@ export function elapsed(start: bigint): string {
return formatDuration(Number(process.hrtime.bigint() - start))
}

export function idTag(uploadId: number, batchId: number): string {
return `[${uploadId}/${batchId}]`
}

const requestTimings = new WeakMap<Request, bigint>()

export const elysiaLogger = new Elysia({ name: 'elysia-logger' })
Expand Down
9 changes: 5 additions & 4 deletions backend/src/handlers/mapillary.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -108,7 +108,7 @@ async function getSequenceIds(sequenceId: string): Promise<string[]> {
return ids
}

async function fetchImagesByIds(imageIds: string[]): Promise<MapillaryImage[]> {
async function fetchImagesByIds(imageIds: string[], tag = ''): Promise<MapillaryImage[]> {
if (imageIds.length === 0) return []

const url = new URL('https://graph.mapillary.com')
Expand All @@ -123,7 +123,8 @@ async function fetchImagesByIds(imageIds: string[]): Promise<MapillaryImage[]> {

const data = (await res.json()) as Record<string, MapillaryImage>
const images = Object.values(data)
logger.debug(`[mapillary] fetched ${images.length}/${imageIds.length} images by ids`)
const prefix = tag ? `${tag} ` : ''
logger.debug(`[mapillary] ${prefix}fetched ${images.length}/${imageIds.length} images by ids`)
return images
}

Expand Down Expand Up @@ -249,8 +250,8 @@ export class MapillaryHandler {
return getSequenceIds(input)
}

async fetchImagesBatch(imageIds: string[], _input: string): Promise<MediaImage[]> {
const raw = await fetchImagesByIds(imageIds)
async fetchImagesBatch(imageIds: string[], _input: string, tag = ''): Promise<MediaImage[]> {
const raw = await fetchImagesByIds(imageIds, tag)
return raw.map(fromMapillary).filter((i): i is MediaImage => i !== null)
}

Expand Down
34 changes: 18 additions & 16 deletions backend/src/mediawiki/client.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@ import {
SourceCdnError,
StorageError,
} from '@backend/core/errors'
import { elapsed, logger } from '@backend/core/logger'
import { elapsed, idTag, logger } from '@backend/core/logger'
import { buildAuthHeader } from '@backend/core/oauthClient'
import { withCsrfTokenRetry } from '@backend/mediawiki/tokenRetry'
import type { Redis } from 'ioredis'
Expand Down Expand Up @@ -238,7 +238,8 @@ export class MediaWikiClient {
const sha1 = createHash('sha1').update(buffer).digest('hex')
const totalChunks = Math.ceil(buffer.length / CHUNK_SIZE)
const start = process.hrtime.bigint()
logger.info(`[mw] uploading ${filename} (${buffer.length} bytes, ${totalChunks} chunks)`)
const tag = idTag(uploadId, batchId)
logger.info(`[mw] ${tag} uploading ${filename} (${buffer.length} bytes, ${totalChunks} chunks)`)

const duplicates = await this.findDuplicates(sha1)
if (duplicates.length > 0) {
Expand Down Expand Up @@ -275,7 +276,7 @@ export class MediaWikiClient {
}

const chunkResponse = await withCsrfTokenRetry(
`[mw] uploadFile chunk ${offset / CHUNK_SIZE + 1}/${totalChunks}`,
`[mw] ${tag} uploadFile chunk ${offset / CHUNK_SIZE + 1}/${totalChunks}`,
() => this.getCsrfToken(),
(t) => this.apiUploadChunk(buildFormData(t)),
token,
Expand All @@ -300,10 +301,10 @@ export class MediaWikiClient {
const upload = result.upload as Record<string, unknown>
stashKey = upload.filekey as string
const chunkNum = offset / CHUNK_SIZE + 1
logger.info(`[mw] chunk ${chunkNum}/${totalChunks} uploaded`)
logger.info(`[mw] ${tag} chunk ${chunkNum}/${totalChunks} uploaded`)
}

logger.info(`[mw] final commit for ${filename}`)
logger.info(`[mw] ${tag} final commit for ${filename}`)
let commitResult: Record<string, unknown> | null = null
for (let attempt = 0; attempt <= STASH_RETRY_LIMIT; attempt++) {
const buildFormData = (t: string) => {
Expand All @@ -318,7 +319,7 @@ export class MediaWikiClient {
}

const commitResponse = await withCsrfTokenRetry(
`[mw] uploadFile commit (attempt ${attempt + 1})`,
`[mw] ${tag} uploadFile commit (attempt ${attempt + 1})`,
() => this.getCsrfToken(),
(t) => this.apiUploadChunk(buildFormData(t), true),
token,
Expand All @@ -329,7 +330,7 @@ export class MediaWikiClient {
if (errorObj) {
if (errorObj.code === 'uploadstash-file-not-found' && attempt < STASH_RETRY_LIMIT) {
logger.warn(
`[mw] stash file not found on attempt ${attempt + 1}, retrying in ${STASH_RETRY_DELAY_MS}ms`,
`[mw] ${tag} stash file not found on attempt ${attempt + 1}, retrying in ${STASH_RETRY_DELAY_MS}ms`,
)
await new Promise((resolve) => setTimeout(resolve, STASH_RETRY_DELAY_MS))
continue
Expand Down Expand Up @@ -364,8 +365,8 @@ export class MediaWikiClient {
)
}
logger.warn(
{ uploadId, filename, result: upload.result, warnings },
`[mw] unexpected upload result during commit`,
{ filename, result: upload.result, warnings },
`[mw] ${tag} unexpected upload result during commit`,
)
throw new Error(`Unexpected upload result: ${upload.result}`)
}
Expand All @@ -374,7 +375,7 @@ export class MediaWikiClient {
const upload = commitResult.upload as Record<string, unknown>
const imageinfo = upload.imageinfo as Record<string, string>
await redis.del(lockKey)
logger.info(`[mw] uploaded ${filename} | ${elapsed(start)}`)
logger.info(`[mw] ${tag} uploaded ${filename} | ${elapsed(start)}`)
return imageinfo.descriptionurl!
} catch (err) {
await redis.del(lockKey)
Expand All @@ -387,13 +388,14 @@ export class MediaWikiClient {
claims: unknown[] | null,
labels: Record<string, { language: string; value: string }> | null,
editSummary: string,
tag: string,
): Promise<void> {
const start = process.hrtime.bigint()
const payload: Record<string, unknown> = {}
if (claims) payload.claims = claims
if (labels) payload.labels = labels
const { result } = await withCsrfTokenRetry(
`[mw] applySdc ${filename}`,
`[mw] ${tag} applySdc ${filename}`,
() => this.getCsrfToken(),
(token) =>
this.apiRequest(
Expand All @@ -404,10 +406,10 @@ export class MediaWikiClient {
)
if (result.error)
throw new Error((result.error as Record<string, string>).info ?? 'wbeditentity failed')
logger.info(`[mw] sdc applied to ${filename} | ${elapsed(start)}`)
logger.info(`[mw] ${tag} sdc applied to ${filename} | ${elapsed(start)}`)
}

async nullEdit(filename: string): Promise<void> {
async nullEdit(filename: string, tag: string): Promise<void> {
const start = process.hrtime.bigint()
for (let attempt = 0; attempt <= STASH_RETRY_LIMIT; attempt++) {
try {
Expand All @@ -425,7 +427,7 @@ export class MediaWikiClient {
const mainSlot = (slots.main as Record<string, unknown>) ?? {}
const content = (mainSlot.content as string) ?? ''
const { result } = await withCsrfTokenRetry(
`[mw] nullEdit ${filename}`,
`[mw] ${tag} nullEdit ${filename}`,
() => this.getCsrfToken(),
(token) =>
this.apiRequest({ action: 'edit' }, 'POST', {
Expand All @@ -438,12 +440,12 @@ export class MediaWikiClient {
)
if (result.error)
throw new Error((result.error as Record<string, string>).info ?? 'null edit failed')
logger.info(`[mw] null edit on ${filename} | ${elapsed(start)}`)
logger.info(`[mw] ${tag} null edit on ${filename} | ${elapsed(start)}`)
return
} catch (err) {
if (attempt < STASH_RETRY_LIMIT) {
logger.warn(
`[mw] null edit attempt ${attempt + 1} failed for ${filename}, retrying in ${STASH_RETRY_DELAY_MS}ms`,
`[mw] ${tag} null edit attempt ${attempt + 1} failed for ${filename}, retrying in ${STASH_RETRY_DELAY_MS}ms`,
)
await new Promise((resolve) => setTimeout(resolve, STASH_RETRY_DELAY_MS))
} else {
Expand Down
8 changes: 5 additions & 3 deletions backend/src/workers/queue.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
import { config } from '@backend/config'
import { logger } from '@backend/core/logger'
import { idTag, logger } from '@backend/core/logger'
import type { RateLimitInfo } from '@backend/core/rateLimiter'
import { Queue } from 'bullmq'

Expand Down Expand Up @@ -42,7 +42,7 @@ export async function enqueueUpload(data: UploadJobData, delayMs: number): Promi
removeOnFail: { age: 86400 * 7 },
})
logger.info(
`[worker] upload ${data.uploadId} enqueued (job: ${job.id}, batch: ${data.batchId}, delay: ${formatDelayMs(delayMs)})`,
`[worker] ${idTag(data.uploadId, data.batchId)} enqueued (job: ${job.id}, delay: ${formatDelayMs(delayMs)})`,
)
return job.id!
}
Expand All @@ -51,6 +51,8 @@ export async function removeUploadJob(jobId: string): Promise<void> {
const job = await getUploadQueue().getJob(jobId)
if (job) {
await job.remove()
logger.info(`[worker] job ${jobId} removed from queue`)
logger.info(
`[worker] ${idTag(job.data.uploadId, job.data.batchId)} job ${jobId} removed from queue`,
)
Comment on lines +54 to +56

@greptile-apps greptile-apps Bot Jul 5, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Removal Log Reads Job Data

When removeUploadJob removes an existing BullMQ job whose stored data is missing or from an older shape, the new log line reads job.data.uploadId and job.data.batchId after the remove. The old path could remove the job using only jobId; this path can now reject during cleanup or emit [undefined/undefined], which can break callers that cancel or requeue stale jobs.

}
}
Loading