From 832f83091ae47be563ee51cbb8135e12255995d8 Mon Sep 17 00:00:00 2001 From: DaxServer Date: Sun, 5 Jul 2026 14:05:10 +0200 Subject: [PATCH] feat: add consistent uploadId/batchId log prefix on worker path Co-Authored-By: Claude Sonnet 5 --- backend/src/__tests__/uploadClient.test.ts | 14 ++-- backend/src/core/logger.ts | 4 + backend/src/handlers/mapillary.ts | 9 ++- backend/src/mediawiki/client.ts | 34 +++++---- backend/src/workers/queue.ts | 8 +- backend/src/workers/upload.worker.ts | 89 ++++++++++------------ 6 files changed, 80 insertions(+), 78 deletions(-) diff --git a/backend/src/__tests__/uploadClient.test.ts b/backend/src/__tests__/uploadClient.test.ts index aa2193c..8c28e14 100644 --- a/backend/src/__tests__/uploadClient.test.ts +++ b/backend/src/__tests__/uploadClient.test.ts @@ -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() }) @@ -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 () => { @@ -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) }) @@ -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 @@ -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 } @@ -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) }) diff --git a/backend/src/core/logger.ts b/backend/src/core/logger.ts index 582d3d7..a7ceb9e 100644 --- a/backend/src/core/logger.ts +++ b/backend/src/core/logger.ts @@ -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() export const elysiaLogger = new Elysia({ name: 'elysia-logger' }) diff --git a/backend/src/handlers/mapillary.ts b/backend/src/handlers/mapillary.ts index f0402b6..34c7efb 100644 --- a/backend/src/handlers/mapillary.ts +++ b/backend/src/handlers/mapillary.ts @@ -108,7 +108,7 @@ async function getSequenceIds(sequenceId: string): Promise { return ids } -async function fetchImagesByIds(imageIds: string[]): Promise { +async function fetchImagesByIds(imageIds: string[], tag = ''): Promise { if (imageIds.length === 0) return [] const url = new URL('https://graph.mapillary.com') @@ -123,7 +123,8 @@ async function fetchImagesByIds(imageIds: string[]): Promise { const data = (await res.json()) as Record 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 } @@ -249,8 +250,8 @@ export class MapillaryHandler { return getSequenceIds(input) } - async fetchImagesBatch(imageIds: string[], _input: string): Promise { - const raw = await fetchImagesByIds(imageIds) + async fetchImagesBatch(imageIds: string[], _input: string, tag = ''): Promise { + const raw = await fetchImagesByIds(imageIds, tag) return raw.map(fromMapillary).filter((i): i is MediaImage => i !== null) } diff --git a/backend/src/mediawiki/client.ts b/backend/src/mediawiki/client.ts index d1f0fa3..f460b25 100644 --- a/backend/src/mediawiki/client.ts +++ b/backend/src/mediawiki/client.ts @@ -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' @@ -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) { @@ -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, @@ -300,10 +301,10 @@ export class MediaWikiClient { const upload = result.upload as Record 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 | null = null for (let attempt = 0; attempt <= STASH_RETRY_LIMIT; attempt++) { const buildFormData = (t: string) => { @@ -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, @@ -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 @@ -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}`) } @@ -374,7 +375,7 @@ export class MediaWikiClient { const upload = commitResult.upload as Record const imageinfo = upload.imageinfo as Record 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) @@ -387,13 +388,14 @@ export class MediaWikiClient { claims: unknown[] | null, labels: Record | null, editSummary: string, + tag: string, ): Promise { const start = process.hrtime.bigint() const payload: Record = {} 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( @@ -404,10 +406,10 @@ export class MediaWikiClient { ) if (result.error) throw new Error((result.error as Record).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 { + async nullEdit(filename: string, tag: string): Promise { const start = process.hrtime.bigint() for (let attempt = 0; attempt <= STASH_RETRY_LIMIT; attempt++) { try { @@ -425,7 +427,7 @@ export class MediaWikiClient { const mainSlot = (slots.main as Record) ?? {} 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', { @@ -438,12 +440,12 @@ export class MediaWikiClient { ) if (result.error) throw new Error((result.error as Record).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 { diff --git a/backend/src/workers/queue.ts b/backend/src/workers/queue.ts index 7882c50..f5b2afc 100644 --- a/backend/src/workers/queue.ts +++ b/backend/src/workers/queue.ts @@ -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' @@ -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! } @@ -51,6 +51,8 @@ export async function removeUploadJob(jobId: string): Promise { 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`, + ) } } diff --git a/backend/src/workers/upload.worker.ts b/backend/src/workers/upload.worker.ts index 504d09a..92ceb6e 100644 --- a/backend/src/workers/upload.worker.ts +++ b/backend/src/workers/upload.worker.ts @@ -7,7 +7,7 @@ import { SourceCdnError, StorageError, } from '@backend/core/errors' -import { logger } from '@backend/core/logger' +import { idTag, logger } from '@backend/core/logger' import type { RateLimitInfo } from '@backend/core/rateLimiter' import { getNextUploadDelay as defaultGetNextUploadDelay } from '@backend/core/rateLimiter' import { lazyDb } from '@backend/db/client' @@ -56,24 +56,22 @@ export function createUploadWorker(redis: Redis, deps?: WorkerDeps): Worker { const { uploadId, batchId, editGroupId } = job.data - logger.info(`[worker] [${uploadId}/${batchId}] task started (job: ${job.id})`) + const tag = idTag(uploadId, batchId) + logger.info(`[worker] ${tag} task started (job: ${job.id})`) const upload = await uploads.getUploadById(uploadId) if (!upload) { - logger.error( - { uploadId, batchId }, - `[worker] [${uploadId}/${batchId}] upload not found, skipping`, - ) + logger.error(`[worker] ${tag} upload not found, skipping`) return } if (upload.status === 'cancelled') { - logger.info(`[worker] [${uploadId}/${batchId}] skipping cancelled upload`) + logger.info(`[worker] ${tag} skipping cancelled upload`) return } if (!upload.access_token) { - logger.warn(`[worker] [${uploadId}/${batchId}] no access token, marking failed`) + logger.warn(`[worker] ${tag} no access token, marking failed`) await uploads.updateUploadStatus(uploadId, 'failed', { type: 'error', message: 'Your session has expired. Please log in and retry.', @@ -85,7 +83,7 @@ export function createUploadWorker(redis: Redis, deps?: WorkerDeps): Worker i.id === upload.key) if (!image) { - logger.warn( - `[worker] [${uploadId}/${batchId}] image ${upload.key} not found in mapillary, will retry`, - ) + logger.warn(`[worker] ${tag} image ${upload.key} not found in mapillary, will retry`) throw new Error(`Image ${upload.key} not found in Mapillary — will retry`) } - logger.info(`[worker] [${uploadId}/${batchId}] starting upload: ${upload.filename}`) + logger.info(`[worker] ${tag} starting upload: ${upload.filename}`) await uploads.updateUploadStatus(uploadId, 'in_progress') const summary = buildEditSummary(upload.key, batchId, editGroupId) @@ -141,21 +139,21 @@ export function createUploadWorker(redis: Redis, deps?: WorkerDeps): Worker 0) { @@ -173,9 +171,7 @@ export function createUploadWorker(redis: Redis, deps?: WorkerDeps): Worker 0 ? claimsDelta : null, labelsDelta, summary, + tag, ) await uploads.updateUploadStatus(uploadId, 'duplicated_sdc_updated', { type: 'duplicated_sdc_updated', @@ -195,18 +192,15 @@ export function createUploadWorker(redis: Redis, deps?: WorkerDeps): Worker { - const uploadId = job?.data.uploadId - const batchId = job?.data.batchId if (!job) return + const { uploadId, batchId } = job.data + const tag = idTag(uploadId, batchId) // BullMQ fires 'failed' on every attempt, not just the final one. // For errors that rely on BullMQ's built-in retry (not StorageError's custom requeue), // reset status to queued on intermediate attempts — the access token must stay intact for retries. if (!(err instanceof StorageError) && job.attemptsMade < (job.opts.attempts ?? 1)) { - logger.warn( - { jobId: job.id, err }, - `[worker] [${uploadId}/${batchId}] job attempt failed, will retry`, - ) + logger.warn({ jobId: job.id, err }, `[worker] ${tag} job attempt failed, will retry`) try { await uploads.updateUploadStatus(job.data.uploadId, 'queued') } catch (dbErr) { logger.error( { jobId: job.id, err: dbErr }, - `[worker] [${uploadId}/${batchId}] failed to update db status after job failure`, + `[worker] ${tag} failed to update db status after job failure`, ) } return } - logger.error({ jobId: job.id, err }, `[worker] [${uploadId}/${batchId}] job permanently failed`) + logger.error({ jobId: job.id, err }, `[worker] ${tag} job permanently failed`) try { if (err instanceof StorageError) { const count = job.data.requeueCount ?? 0 if (count >= MAX_STORAGE_REQUEUE_COUNT) { logger.warn( - `[worker] [${uploadId}/${batchId}] StorageError requeue limit reached (${count}), marking failed`, + `[worker] ${tag} StorageError requeue limit reached (${count}), marking failed`, ) } else { await uploads.updateUploadStatus(job.data.uploadId, 'queued') await removeUploadJobFn(job.id!) const delay = await getNextUploadDelayFn(job.data.userid, job.data.rateLimit, redis) logger.info( - `[worker] [${uploadId}/${batchId}] requeuing after StorageError (attempt ${count + 1}/${MAX_STORAGE_REQUEUE_COUNT}, delay: ${formatDelayMs(delay)})`, + `[worker] ${tag} requeuing after StorageError (attempt ${count + 1}/${MAX_STORAGE_REQUEUE_COUNT}, delay: ${formatDelayMs(delay)})`, ) await enqueueUploadFn({ ...job.data, requeueCount: count + 1 }, delay) return @@ -297,7 +288,7 @@ export function createUploadWorker(redis: Redis, deps?: WorkerDeps): Worker