From a2e6637bd41e0e1094e92d6e726f51f2e6383471 Mon Sep 17 00:00:00 2001 From: Dylan Jeffers Date: Mon, 27 Jul 2026 23:05:13 -0700 Subject: [PATCH] fix(archiver): replace stale jobs so retry can escape a wedged download MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Job ids are deterministic (`base64url("{userId}-{trackId}")`), and getOrCreateStemsArchiveJob only replaced a job whose state was `failed`. Any job parked in a non-terminal state was therefore handed straight back to every subsequent retry from that user for that track. That is what made the 2026-07-28 outage unrecoverable from the client. The worker lost its Redis lock when archiver-redis was recreated, kept all five concurrency slots, and jobs sat in `waiting` — never `failed`. The "Try again" link in DownloadTrackArchiveModal is the only recovery affordance the UI offers, and it was a no-op: POST returned the same dead job id, the client re-polled it, and timed out again 15 minutes later. Cancel-then-retry could not work around it either. cancelStemsArchiveJob removes a job from the queue only when it is completed; otherwise it aborts via an in-memory AbortController, and a job stranded in `waiting` has none, so it falls through to 'Stems archive job not found' and removes nothing. Now a job that has been non-terminal past `staleJobSeconds` (default 15m, `ARCHIVER_STALE_JOB_SECONDS`) is removed and recreated. Completed jobs are still returned as-is so a ready archive is downloaded rather than rebuilt, and jobs that are still progressing are still rejoined, which is what dedupes double-clicks and multiple tabs. The threshold is kept in step with STEMS_ARCHIVE_POLL_TIMEOUT_MS in the client's useDownloadTrackStems: past that point the client has already given up, so the old job's output would never be collected anyway. Staleness is measured from `processedOn ?? timestamp`, so a job that never started — the case that actually broke — is covered. Co-Authored-By: Claude Opus 5 (1M context) --- apps/archiver/src/config.ts | 17 ++ .../src/jobs/createStemsArchive.test.ts | 202 ++++++++++++++++++ apps/archiver/src/jobs/createStemsArchive.ts | 45 +++- 3 files changed, 263 insertions(+), 1 deletion(-) create mode 100644 apps/archiver/src/jobs/createStemsArchive.test.ts diff --git a/apps/archiver/src/config.ts b/apps/archiver/src/config.ts index 0b749aad..e5d3c4af 100644 --- a/apps/archiver/src/config.ts +++ b/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({ 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, diff --git a/apps/archiver/src/jobs/createStemsArchive.test.ts b/apps/archiver/src/jobs/createStemsArchive.test.ts new file mode 100644 index 00000000..f142c865 --- /dev/null +++ b/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 }) + }) +}) diff --git a/apps/archiver/src/jobs/createStemsArchive.ts b/apps/archiver/src/jobs/createStemsArchive.ts index e11e18ca..44ade4b4 100644 --- a/apps/archiver/src/jobs/createStemsArchive.ts +++ b/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 ) => { @@ -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() }