feat: add consistent uploadId/batchId log prefix on worker path (#114)

Co-authored-by: Claude Sonnet 5 <noreply@anthropic.com>

Author: DaxServer

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)
   })

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<Request, bigint>()
 
 export const elysiaLogger = new Elysia({ name: 'elysia-logger' })

backend/src/handlers/mapillary.ts

@@ -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')
@@ -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
 }
 
@@ -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)
   }
 

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<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) => {
@@ -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<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)
@@ -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(
@@ -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 {
@@ -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', {
@@ -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 {

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<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`,
+    )
   }
 }