fix(archiver): replace stale jobs so retry can escape a wedged download

apps/archiver/src/config.ts

@@ -22,6 +22,21 @@ export type Config = {
   concurrentJobs: number
   /** How many attempts to make to create a stems archive (default: 3) */
   maxStemsArchiveAttempts: number
+  /**
+   * How long a stems archive job may sit in a non-terminal state before a new
+   * request for the same track replaces it rather than joining it
+   * (default: 15 minutes).
+   *
+   * Job ids are deterministic, so without this a job that can never finish —
+   * a worker that died or lost its Redis lock still holding the slot — is
+   * handed back to every subsequent retry from that user, and the download is
+   * permanently unrecoverable from the client.
+   *
+   * Kept in step with STEMS_ARCHIVE_POLL_TIMEOUT_MS in the client's
+   * useDownloadTrackStems: past that point the client has already given up on
+   * the job, so its output would never be collected anyway.
+   */
+  staleJobSeconds: number
   redisUrl: string
   serverHost: string
   serverPort: number
@@ -66,6 +81,7 @@ export const readConfig = (): Config => {
     archiver_orphaned_jobs_lifetime_seconds: num({ default: 60 * 10 }),
     archiver_log_level: str<LogLevel>({ default: 'info' }),
     archiver_max_stems_archive_attempts: num({ default: 3 }),
+    archiver_stale_job_seconds: num({ default: 60 * 15 }),
     archiver_max_disk_space_bytes: num({
       default: 32 * 1024 * 1024 * 1024
     }), // 32GB
@@ -84,6 +100,7 @@ export const readConfig = (): Config => {
       env.archiver_cleanup_orphaned_files_interval_seconds,
     orphanedJobsLifetimeSeconds: env.archiver_orphaned_jobs_lifetime_seconds,
     maxStemsArchiveAttempts: env.archiver_max_stems_archive_attempts,
+    staleJobSeconds: env.archiver_stale_job_seconds,
     maxDiskSpaceBytes: env.archiver_max_disk_space_bytes,
     maxDiskSpaceWaitSeconds: env.archiver_max_disk_space_wait_seconds,
     logLevel: env.archiver_log_level,

apps/archiver/src/jobs/createStemsArchive.test.ts

@@ -0,0 +1,202 @@
+import { describe, it, expect, vi, beforeEach } from 'vitest'
+
+const mockQueue = {
+  getJob: vi.fn(),
+  add: vi.fn()
+}
+
+vi.mock('bullmq', () => ({
+  // `new Queue(...)` in getStemsArchiveQueue needs a real constructor;
+  // returning an object from one hands back the shared mock.
+  Queue: class {
+    constructor() {
+      return mockQueue
+    }
+  }
+}))
+
+const STALE_JOB_SECONDS = 900
+
+vi.mock('../config', () => ({
+  readConfig: () => ({
+    redisUrl: 'redis://localhost:6379',
+    maxStemsArchiveAttempts: 3,
+    orphanedJobsLifetimeSeconds: 600,
+    staleJobSeconds: STALE_JOB_SECONDS
+  })
+}))
+
+import {
+  getOrCreateStemsArchiveJob,
+  isStaleJob,
+  generateJobId
+} from './createStemsArchive'
+
+const STALE_MS = STALE_JOB_SECONDS * 1000
+
+const jobData = {
+  trackId: 1595511751,
+  userId: 700900251,
+  messageHeader: 'signature:1785215370374',
+  signatureHeader: '0xdeadbeef',
+  includeParentTrack: false
+}
+
+const makeJob = ({
+  state,
+  timestamp,
+  processedOn
+}: {
+  state: string
+  timestamp?: number
+  processedOn?: number
+}) => ({
+  getState: vi.fn().mockResolvedValue(state),
+  remove: vi.fn().mockResolvedValue(undefined),
+  timestamp,
+  processedOn,
+  progress: 0,
+  failedReason: undefined,
+  returnvalue: undefined
+})
+
+describe('isStaleJob', () => {
+  const now = 1_000_000_000
+
+  it('is false for a job that was picked up recently', () => {
+    expect(
+      isStaleJob({ processedOn: now - 1000, timestamp: now - 5000 }, STALE_MS, now)
+    ).toBe(false)
+  })
+
+  it('is true once the job has been running past the threshold', () => {
+    expect(
+      isStaleJob(
+        { processedOn: now - (STALE_MS + 1), timestamp: now - STALE_MS * 2 },
+        STALE_MS,
+        now
+      )
+    ).toBe(true)
+  })
+
+  it('falls back to queue time for a job that never started', () => {
+    // The wedged-worker case: parked in `waiting`, so no processedOn at all.
+    expect(
+      isStaleJob(
+        { processedOn: undefined, timestamp: now - (STALE_MS + 1) },
+        STALE_MS,
+        now
+      )
+    ).toBe(true)
+    expect(
+      isStaleJob({ processedOn: undefined, timestamp: now - 1000 }, STALE_MS, now)
+    ).toBe(false)
+  })
+
+  it('prefers processedOn over timestamp', () => {
+    // Queued long ago but picked up just now — actively being worked, not stale.
+    expect(
+      isStaleJob(
+        { processedOn: now - 1000, timestamp: now - STALE_MS * 10 },
+        STALE_MS,
+        now
+      )
+    ).toBe(false)
+  })
+
+  it('treats a job with no usable timestamp as fresh', () => {
+    expect(
+      isStaleJob({ processedOn: undefined, timestamp: undefined }, STALE_MS, now)
+    ).toBe(false)
+  })
+})
+
+describe('getOrCreateStemsArchiveJob', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    mockQueue.add.mockImplementation(async () =>
+      makeJob({ state: 'waiting', timestamp: Date.now() })
+    )
+  })
+
+  it('returns a completed job instead of rebuilding the archive', async () => {
+    const existing = makeJob({
+      state: 'completed',
+      timestamp: Date.now() - STALE_MS * 10
+    })
+    mockQueue.getJob.mockResolvedValue(existing)
+
+    const status = await getOrCreateStemsArchiveJob(jobData)
+
+    expect(existing.remove).not.toHaveBeenCalled()
+    expect(mockQueue.add).not.toHaveBeenCalled()
+    expect(status.state).toBe('completed')
+  })
+
+  it('rejoins an in-flight job that is still making progress', async () => {
+    const existing = makeJob({
+      state: 'active',
+      timestamp: Date.now() - 5000,
+      processedOn: Date.now() - 5000
+    })
+    mockQueue.getJob.mockResolvedValue(existing)
+
+    await getOrCreateStemsArchiveJob(jobData)
+
+    // Dedupes double-clicks and multiple tabs onto the same job.
+    expect(existing.remove).not.toHaveBeenCalled()
+    expect(mockQueue.add).not.toHaveBeenCalled()
+  })
+
+  it('replaces a job wedged in waiting past the stale threshold', async () => {
+    // The 2026-07-28 outage: worker lost its Redis lock, held every slot, and
+    // jobs sat in `waiting` forever. Retry used to be handed this same job.
+    const existing = makeJob({
+      state: 'waiting',
+      timestamp: Date.now() - (STALE_MS + 60_000)
+    })
+    mockQueue.getJob.mockResolvedValue(existing)
+
+    await getOrCreateStemsArchiveJob(jobData)
+
+    expect(existing.remove).toHaveBeenCalledOnce()
+    expect(mockQueue.add).toHaveBeenCalledOnce()
+  })
+
+  it('replaces a job stuck in active past the stale threshold', async () => {
+    const existing = makeJob({
+      state: 'active',
+      timestamp: Date.now() - STALE_MS * 3,
+      processedOn: Date.now() - (STALE_MS + 1000)
+    })
+    mockQueue.getJob.mockResolvedValue(existing)
+
+    await getOrCreateStemsArchiveJob(jobData)
+
+    expect(existing.remove).toHaveBeenCalledOnce()
+    expect(mockQueue.add).toHaveBeenCalledOnce()
+  })
+
+  it('still replaces a failed job regardless of age', async () => {
+    const existing = makeJob({ state: 'failed', timestamp: Date.now() })
+    mockQueue.getJob.mockResolvedValue(existing)
+
+    await getOrCreateStemsArchiveJob(jobData)
+
+    expect(existing.remove).toHaveBeenCalledOnce()
+    expect(mockQueue.add).toHaveBeenCalledOnce()
+  })
+
+  it('creates a job when none exists, keyed by the deterministic id', async () => {
+    mockQueue.getJob.mockResolvedValue(null)
+
+    await getOrCreateStemsArchiveJob(jobData)
+
+    const expectedId = generateJobId({
+      userId: jobData.userId,
+      trackId: jobData.trackId
+    })
+    expect(mockQueue.add).toHaveBeenCalledOnce()
+    expect(mockQueue.add.mock.calls[0][2]).toMatchObject({ jobId: expectedId })
+  })
+})

apps/archiver/src/jobs/createStemsArchive.ts

@@ -77,6 +77,30 @@ const getJobStatus = async (
   }
 }
 
+/**
+ * Whether a job has been sitting in a non-terminal state long enough that we
+ * should assume nobody is going to finish it.
+ *
+ * Measured from when the job was last picked up, falling back to when it was
+ * queued — a job wedged in `waiting` never gets a `processedOn` at all, and
+ * that is the case we most need to catch.
+ */
+export const isStaleJob = (
+  // Deliberately wider than bullmq's `Job`, which types `timestamp` as always
+  // present. These come back off a Redis hash that a previous process wrote,
+  // so treat both fields as possibly missing rather than trusting the type.
+  job: { processedOn?: number | null; timestamp?: number | null },
+  staleAfterMs: number,
+  now: number = Date.now()
+): boolean => {
+  const startedAt = job.processedOn ?? job.timestamp
+  if (typeof startedAt !== 'number') {
+    // No usable timestamp — treat as fresh rather than churning a live job.
+    return false
+  }
+  return now - startedAt > staleAfterMs
+}
+
 export const getOrCreateStemsArchiveJob = async (
   data: Omit<StemsArchiveJobData, 'jobId'>
 ) => {
@@ -87,9 +111,28 @@ export const getOrCreateStemsArchiveJob = async (
   const existingJob = await queue.getJob(jobId)
   if (existingJob) {
     const state = await existingJob.getState()
-    if (state !== 'failed') {
+
+    // Completed means the archive is on disk waiting to be collected — hand
+    // it straight back so the client downloads it instead of rebuilding it.
+    if (state === 'completed') {
       return getJobStatus(jobId, existingJob)
     }
+
+    // Job ids are deterministic (`{userId}-{trackId}`), so a user retrying the
+    // same track always lands on their existing job. Rejoining it is right
+    // while it's making progress — that's what dedupes double-clicks and
+    // multiple tabs — but it's a trap once the job can no longer finish. A
+    // worker that died or lost its Redis lock leaves the job parked in
+    // `waiting`/`active` forever, and because that isn't `failed`, every
+    // retry used to be handed the same dead job. Retry was a no-op and the
+    // download was unrecoverable from the client. Replace it instead.
+    if (
+      state !== 'failed' &&
+      !isStaleJob(existingJob, config.staleJobSeconds * 1000)
+    ) {
+      return getJobStatus(jobId, existingJob)
+    }
+
     await existingJob.remove()
   }