From 3c85acbdd89075e58b2c81a1dca588e990a4887c Mon Sep 17 00:00:00 2001 From: crisog Date: Sat, 28 Feb 2026 20:57:16 -0400 Subject: [PATCH 1/7] feat: implement stateless binary response caching --- .env.example | 7 + README.md | 19 ++ apps/server/package.json | 4 + apps/server/src/config/env.ts | 29 +- apps/server/src/server.ts | 16 ++ apps/server/src/test-utils/test-helpers.ts | 53 ++++ apps/server/src/utils/cache.test.ts | 138 ++++++++++ apps/server/src/utils/cache.ts | 297 +++++++++++++++++++++ apps/server/src/utils/job-handler.test.ts | 227 +++++++++++++++- apps/server/src/utils/job-handler.ts | 80 ++++-- package-lock.json | 249 ++++++++++++++++- 11 files changed, 1080 insertions(+), 39 deletions(-) create mode 100644 apps/server/src/test-utils/test-helpers.ts create mode 100644 apps/server/src/utils/cache.test.ts create mode 100644 apps/server/src/utils/cache.ts diff --git a/.env.example b/.env.example index 3b70cb9..0154385 100644 --- a/.env.example +++ b/.env.example @@ -13,6 +13,13 @@ WORKER_CONCURRENCY=5 # Storage Mode (stateless or s3) STORAGE_MODE=stateless +# Stateless binary response cache (cacache) +CACHE_ENABLED=false +CACHE_DIR=/tmp/ffmpeg-rest/cache +CACHE_TTL_HOURS=2160 +CACHE_MAX_SIZE_MB=1024 +CACHE_SWEEP_INTERVAL_MINUTES=60 + S3_ENDPOINT=https://s3.amazonaws.com S3_REGION=us-east-1 S3_BUCKET=my-bucket diff --git a/README.md b/README.md index 2e6446a..e902ec7 100644 --- a/README.md +++ b/README.md @@ -58,6 +58,25 @@ Files are processed and returned directly in the HTTP response. Simple and strai **Cost Consideration**: On Railway, stateless mode is cheaper than running S3 Mode unless you have free egress at your S3-storage provider (like Cloudflare R2). Railway charges $0.05 per GB egress vs S3's typical $0.09 per GB, but you trade off file persistence - processed files aren't stored for later retrieval. +#### Stateless Binary Cache + +Stateless mode can optionally cache binary conversion outputs using `cacache` to avoid rerunning FFmpeg on identical inputs + params. + +- Cache scope: binary conversion endpoints only (not `/.../url` S3 responses, not `/media/info`) +- Cache key: SHA-256 of input bytes + job type + normalized processing params +- Retention: TTL + size cap with periodic cleanup +- Storage: local filesystem (ephemeral by default) + +**Configuration**: + +```bash +CACHE_ENABLED=false # Enable/disable stateless cache +CACHE_DIR=/tmp/ffmpeg-rest/cache +CACHE_TTL_HOURS=2160 # 90 days +CACHE_MAX_SIZE_MB=1024 # 1 GiB +CACHE_SWEEP_INTERVAL_MINUTES=60 +``` + ### S3 Mode Processed files are uploaded to S3-compatible storage and a URL is returned. This mode significantly reduces egress bandwidth costs since users download the processed files directly from S3 rather than through your API server. Ideal for production deployments where bandwidth costs matter. diff --git a/apps/server/package.json b/apps/server/package.json index 2440b79..508e260 100644 --- a/apps/server/package.json +++ b/apps/server/package.json @@ -18,11 +18,15 @@ "@scalar/hono-api-reference": "^0.9.20", "@scalar/openapi-to-markdown": "^0.2.43", "bullmq": "^5.60.0", + "cacache": "^20.0.3", "dotenv": "^17.2.3", "hono": "^4.9.9", "ioredis": "^5.8.0", "pino": "^10.0.0", "pino-pretty": "^13.1.1", "zod": "^4.1.11" + }, + "devDependencies": { + "@types/cacache": "^20.0.1" } } diff --git a/apps/server/src/config/env.ts b/apps/server/src/config/env.ts index 6f34edc..8482aa8 100644 --- a/apps/server/src/config/env.ts +++ b/apps/server/src/config/env.ts @@ -1,10 +1,27 @@ import { z } from 'zod'; +import path from 'path'; if (process.env['NODE_ENV'] !== 'production') { const dotenv = await import('dotenv'); dotenv.config(); } +const EnvBooleanSchema = z.preprocess((value) => { + if (typeof value !== 'string') { + return value; + } + + const normalized = value.trim().toLowerCase(); + if (['1', 'true', 'yes', 'on'].includes(normalized)) { + return true; + } + if (['0', 'false', 'no', 'off', ''].includes(normalized)) { + return false; + } + + return value; +}, z.boolean()); + const schema = z.object({ PORT: z.coerce.number().default(3000), NODE_ENV: z.enum(['development', 'production', 'test']).default('development'), @@ -15,8 +32,18 @@ const schema = z.object({ MAX_FILE_SIZE: z.coerce.number().default(100 * 1024 * 1024), STORAGE_MODE: z.enum(['stateless', 's3']).default('stateless'), + CACHE_ENABLED: EnvBooleanSchema.default(false), + CACHE_DIR: z.string().optional(), + CACHE_TTL_HOURS: z.coerce.number().int().positive().default(2160), + CACHE_MAX_SIZE_MB: z.coerce.number().int().positive().default(1024), + CACHE_SWEEP_INTERVAL_MINUTES: z.coerce.number().int().positive().default(60), AUTH_TOKEN: z.string().optional() }); -export const env = schema.parse(process.env); +const parsedEnv = schema.parse(process.env); + +export const env = { + ...parsedEnv, + CACHE_DIR: parsedEnv.CACHE_DIR ?? path.join(parsedEnv.TEMP_DIR, 'cache') +}; diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index e385071..254a815 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -3,9 +3,25 @@ import { createApp } from '~/app'; import { env } from '~/config/env'; import { checkRedisHealth } from '~/config/redis'; import { logger } from '~/config/logger'; +import { initCacheDir, startCacheCleanup } from '~/utils/cache'; await checkRedisHealth(); +let stopCacheCleanup: () => void = () => undefined; +if (env.CACHE_ENABLED) { + await initCacheDir(); + stopCacheCleanup = startCacheCleanup(); +} + +const handleShutdown = (signal: NodeJS.Signals) => { + logger.info({ signal }, 'Shutting down server'); + stopCacheCleanup(); + process.exit(0); +}; + +process.once('SIGINT', () => handleShutdown('SIGINT')); +process.once('SIGTERM', () => handleShutdown('SIGTERM')); + const app = createApp(); serve( diff --git a/apps/server/src/test-utils/test-helpers.ts b/apps/server/src/test-utils/test-helpers.ts new file mode 100644 index 0000000..368b2b5 --- /dev/null +++ b/apps/server/src/test-utils/test-helpers.ts @@ -0,0 +1,53 @@ +import { mkdtemp, rm } from 'fs/promises'; +import path from 'path'; +import { tmpdir } from 'os'; + +interface CacheTestEnvOptions { + cacheDir: string; + tempDir: string; + cacheEnabled?: boolean; + ttlHours?: number; + maxSizeMb?: number; + sweepIntervalMinutes?: number; +} + +export function createTempDirTracker() { + const createdDirs: string[] = []; + + return { + async createTempDir(prefix: string): Promise { + const dir = await mkdtemp(path.join(tmpdir(), prefix)); + createdDirs.push(dir); + return dir; + }, + async cleanupTempDirs(): Promise { + await Promise.all( + createdDirs.splice(0, createdDirs.length).map((dir) => + rm(dir, { + recursive: true, + force: true + }) + ) + ); + } + }; +} + +export function setCacheTestEnv(options: CacheTestEnvOptions): void { + process.env['NODE_ENV'] = 'test'; + process.env['TEMP_DIR'] = options.tempDir; + process.env['CACHE_ENABLED'] = options.cacheEnabled === false ? 'false' : 'true'; + process.env['CACHE_DIR'] = options.cacheDir; + process.env['CACHE_TTL_HOURS'] = String(options.ttlHours ?? 24); + process.env['CACHE_MAX_SIZE_MB'] = String(options.maxSizeMb ?? 10); + process.env['CACHE_SWEEP_INTERVAL_MINUTES'] = String(options.sweepIntervalMinutes ?? 60); +} + +export function clearCacheTestEnv(): void { + delete process.env['CACHE_ENABLED']; + delete process.env['CACHE_DIR']; + delete process.env['CACHE_TTL_HOURS']; + delete process.env['CACHE_MAX_SIZE_MB']; + delete process.env['CACHE_SWEEP_INTERVAL_MINUTES']; + delete process.env['TEMP_DIR']; +} diff --git a/apps/server/src/utils/cache.test.ts b/apps/server/src/utils/cache.test.ts new file mode 100644 index 0000000..df1811e --- /dev/null +++ b/apps/server/src/utils/cache.test.ts @@ -0,0 +1,138 @@ +import { describe, it, expect, afterEach, vi } from 'vitest'; +import * as cacache from 'cacache'; +import { clearCacheTestEnv, createTempDirTracker, setCacheTestEnv } from '../test-utils/test-helpers'; + +const { createTempDir, cleanupTempDirs } = createTempDirTracker(); + +async function loadCacheModule(options?: { + cacheEnabled?: boolean; + ttlHours?: number; + maxSizeMb?: number; + sweepIntervalMinutes?: number; +}) { + const cacheDir = await createTempDir('cache-utils-'); + const tempDir = await createTempDir('cache-utils-temp-'); + + vi.resetModules(); + vi.clearAllMocks(); + + setCacheTestEnv({ + tempDir, + cacheDir, + cacheEnabled: options?.cacheEnabled, + ttlHours: options?.ttlHours, + maxSizeMb: options?.maxSizeMb, + sweepIntervalMinutes: options?.sweepIntervalMinutes + }); + + const mod = await import('./cache'); + return { ...mod, cacheDir }; +} + +afterEach(async () => { + await cleanupTempDirs(); + clearCacheTestEnv(); +}); + +describe('cache utility', () => { + it('should generate deterministic cache keys for identical input', async () => { + const { computeCacheKey } = await loadCacheModule(); + const input = Buffer.from('same-input'); + const params = { quality: 2, mode: 'fit' }; + + const key1 = computeCacheKey(input, 'audio:mp3', params); + const key2 = computeCacheKey(input, 'audio:mp3', params); + + expect(key1).toBe(key2); + }); + + it('should generate different keys for different params and job types', async () => { + const { computeCacheKey } = await loadCacheModule(); + const input = Buffer.from('same-input'); + + const keyA = computeCacheKey(input, 'audio:mp3', { quality: 2 }); + const keyB = computeCacheKey(input, 'audio:mp3', { quality: 7 }); + const keyC = computeCacheKey(input, 'video:mp4', { quality: 2 }); + + expect(keyA).not.toBe(keyB); + expect(keyA).not.toBe(keyC); + }); + + it('should strip runtime path and S3 keys from cacheable params', async () => { + const { extractCacheableParams } = await loadCacheModule(); + const params = extractCacheableParams({ + inputPath: '/tmp/input', + outputPath: '/tmp/output', + outputDir: '/tmp/frames', + jobDir: '/tmp/job', + uploadToS3: true, + quality: 2, + nested: { + outputPath: '/tmp/nested', + mode: 'fit' + } + }); + + expect(params).toEqual({ + nested: { mode: 'fit' }, + quality: 2 + }); + }); + + it('should round-trip cached output', async () => { + const { initCacheDir, putCachedOutput, getCachedOutput } = await loadCacheModule(); + await initCacheDir(); + + const key = 'roundtrip-key'; + const output = Buffer.from('converted-output'); + await putCachedOutput(key, output, 'audio:mp3', 'mp3', { codec: 'mp3' }); + + const cached = await getCachedOutput(key); + expect(cached?.outputBuffer.toString()).toBe('converted-output'); + expect(cached?.metadata).toEqual({ codec: 'mp3' }); + }); + + it('should treat expired entries as cache misses', async () => { + const { initCacheDir, getCachedOutput, cacheDir } = await loadCacheModule({ ttlHours: 1 }); + await initCacheDir(); + + const key = 'expired-key'; + await cacache.put(cacheDir, key, Buffer.from('old-data'), { + metadata: { + createdAt: Date.now() - 2 * 60 * 60 * 1000, + jobType: 'audio:mp3', + outputExtension: 'mp3' + } + }); + + const cached = await getCachedOutput(key); + expect(cached).toBeNull(); + }); + + it('should evict oldest entries when cache exceeds max size', async () => { + const { initCacheDir, putCachedOutput, getCachedOutput } = await loadCacheModule({ maxSizeMb: 1 }); + await initCacheDir(); + + const keyOld = 'old-entry'; + const keyNew = 'new-entry'; + const payload = Buffer.alloc(700 * 1024, 1); + + await putCachedOutput(keyOld, payload, 'audio:mp3', 'mp3'); + await new Promise((resolve) => setTimeout(resolve, 10)); + await putCachedOutput(keyNew, payload, 'audio:mp3', 'mp3'); + + const oldEntry = await getCachedOutput(keyOld); + const newEntry = await getCachedOutput(keyNew); + + expect(oldEntry).toBeNull(); + expect(newEntry).not.toBeNull(); + }); + + it('should no-op when cache is disabled', async () => { + const { putCachedOutput, getCachedOutput } = await loadCacheModule({ cacheEnabled: false }); + await putCachedOutput('disabled-key', Buffer.from('data'), 'audio:mp3', 'mp3'); + + const cached = await getCachedOutput('disabled-key'); + expect(cached).toBeNull(); + }); +}); diff --git a/apps/server/src/utils/cache.ts b/apps/server/src/utils/cache.ts new file mode 100644 index 0000000..0dbeb91 --- /dev/null +++ b/apps/server/src/utils/cache.ts @@ -0,0 +1,297 @@ +import { createHash } from 'crypto'; +import { mkdir } from 'fs/promises'; +import * as cacache from 'cacache'; +import { z } from 'zod'; +import { env } from '~/config/env'; +import { logger } from '~/config/logger'; + +const EXCLUDED_KEYS = new Set(['inputPath', 'outputPath', 'outputDir', 'jobDir', 'uploadToS3']); + +const CacheMetadataSchema = z.object({ + createdAt: z.number(), + jobType: z.string(), + outputExtension: z.string(), + resultMetadata: z.record(z.string(), z.unknown()).optional() +}); + +type CacheMetadata = z.infer; + +interface CacheHit { + outputBuffer: Buffer; + metadata?: Record; +} + +interface CacheEntryLike { + key: string; + size: number; + time: number; + metadata?: unknown; +} + +function normalizeValue(value: unknown): unknown { + if (value === undefined || typeof value === 'function' || typeof value === 'symbol') { + return undefined; + } + + if (value === null || typeof value === 'string' || typeof value === 'boolean') { + return value; + } + + if (typeof value === 'number') { + return Number.isFinite(value) ? value : null; + } + + if (typeof value === 'bigint') { + return value.toString(); + } + + if (value instanceof Date) { + return value.toISOString(); + } + + if (Buffer.isBuffer(value)) { + return value.toString('base64'); + } + + if (Array.isArray(value)) { + return value.map((item) => { + const normalized = normalizeValue(item); + return normalized === undefined ? null : normalized; + }); + } + + if (typeof value === 'object') { + const record = value as Record; + const normalized: Record = {}; + + for (const key of Object.keys(record).sort()) { + if (EXCLUDED_KEYS.has(key)) { + continue; + } + + const normalizedValue = normalizeValue(record[key]); + if (normalizedValue !== undefined) { + normalized[key] = normalizedValue; + } + } + + return normalized; + } + + return value; +} + +function stableStringify(value: unknown): string { + return JSON.stringify(normalizeValue(value)); +} + +function isMissingCacheEntryError(error: unknown): boolean { + if (!error || typeof error !== 'object') { + return false; + } + + const code = (error as { code?: string }).code; + return code === 'ENOENT' || code === 'ENODATA'; +} + +function getMaxCacheBytes(): number { + return env.CACHE_MAX_SIZE_MB * 1024 * 1024; +} + +function ttlMs(): number { + return env.CACHE_TTL_HOURS * 60 * 60 * 1000; +} + +function parseCacheMetadata(metadata: unknown): CacheMetadata | null { + const parsed = CacheMetadataSchema.safeParse(metadata); + return parsed.success ? parsed.data : null; +} + +function getEntryTimestamp(entry: CacheEntryLike): number { + const metadata = parseCacheMetadata(entry.metadata); + return metadata?.createdAt ?? entry.time; +} + +function isExpiredByTimestamp(timestamp: number): boolean { + return Date.now() - timestamp > ttlMs(); +} + +function isExpiredEntry(entry: CacheEntryLike): boolean { + return isExpiredByTimestamp(getEntryTimestamp(entry)); +} + +async function runVerifyAfterRemovals(hadRemovals: boolean): Promise { + if (!hadRemovals) { + return; + } + + try { + await cacache.verify(env.CACHE_DIR, { concurrency: 8 }); + } catch (error) { + logger.warn({ error }, 'Cache verify failed after retention cleanup'); + } +} + +export function isCacheEligibleJobData(jobDataResult: Record): boolean { + const uploadToS3 = jobDataResult['uploadToS3']; + return uploadToS3 !== true; +} + +export function extractCacheableParams(jobDataResult: Record): Record { + const normalized = normalizeValue(jobDataResult); + if (!normalized || typeof normalized !== 'object' || Array.isArray(normalized)) { + return {}; + } + return normalized as Record; +} + +export function computeCacheKey(fileBuffer: Buffer, jobType: string, params: Record): string { + const hash = createHash('sha256'); + hash.update(fileBuffer); + hash.update('\0'); + hash.update(jobType); + hash.update('\0'); + hash.update(stableStringify(params)); + return hash.digest('hex'); +} + +export async function getCachedOutput(cacheKey: string): Promise { + if (!env.CACHE_ENABLED) { + return null; + } + + try { + const entry = await cacache.get(env.CACHE_DIR, cacheKey, { memoize: false }); + const metadata = parseCacheMetadata(entry.metadata); + + if (!metadata) { + await cacache.rm.entry(env.CACHE_DIR, cacheKey); + logger.warn({ cacheKey }, 'Removed cache entry with invalid metadata'); + return null; + } + + if (isExpiredByTimestamp(metadata.createdAt)) { + await cacache.rm.entry(env.CACHE_DIR, cacheKey); + return null; + } + + return { + outputBuffer: entry.data, + metadata: metadata.resultMetadata + }; + } catch (error) { + if (isMissingCacheEntryError(error)) { + return null; + } + + logger.warn({ error, cacheKey }, 'Failed to read cache entry'); + return null; + } +} + +async function enforceCacheRetention(requiredSpaceBytes = 0): Promise { + if (!env.CACHE_ENABLED) { + return; + } + + const maxBytes = getMaxCacheBytes(); + if (requiredSpaceBytes > maxBytes) { + return; + } + + try { + const entriesMap = await cacache.ls(env.CACHE_DIR); + const entries = Object.values(entriesMap) as CacheEntryLike[]; + let hadRemovals = false; + + const activeEntries: CacheEntryLike[] = []; + + for (const entry of entries) { + if (isExpiredEntry(entry)) { + await cacache.rm.entry(env.CACHE_DIR, entry.key); + hadRemovals = true; + } else { + activeEntries.push(entry); + } + } + + let currentSize = activeEntries.reduce((total, entry) => total + entry.size, 0); + const sortedByAge = activeEntries.sort((a, b) => getEntryTimestamp(a) - getEntryTimestamp(b)); + + while (currentSize + requiredSpaceBytes > maxBytes && sortedByAge.length > 0) { + const oldest = sortedByAge.shift(); + if (!oldest) { + break; + } + + await cacache.rm.entry(env.CACHE_DIR, oldest.key); + currentSize -= oldest.size; + hadRemovals = true; + } + + await runVerifyAfterRemovals(hadRemovals); + } catch (error) { + logger.warn({ error }, 'Failed to enforce cache retention'); + } +} + +export async function putCachedOutput( + cacheKey: string, + outputBuffer: Buffer, + jobType: string, + outputExtension: string, + metadata?: Record +): Promise { + if (!env.CACHE_ENABLED) { + return; + } + + const maxBytes = getMaxCacheBytes(); + if (outputBuffer.length > maxBytes) { + logger.debug({ cacheKey, size: outputBuffer.length }, 'Skipping cache write because output exceeds max cache size'); + return; + } + + try { + await enforceCacheRetention(outputBuffer.length); + + await cacache.put(env.CACHE_DIR, cacheKey, outputBuffer, { + metadata: { + createdAt: Date.now(), + jobType, + outputExtension, + resultMetadata: metadata + } + }); + } catch (error) { + logger.warn({ error, cacheKey }, 'Failed to write cache entry'); + } +} + +export async function initCacheDir(): Promise { + if (!env.CACHE_ENABLED) { + return; + } + + await mkdir(env.CACHE_DIR, { recursive: true }); + await enforceCacheRetention(); +} + +export function startCacheCleanup(): () => void { + if (!env.CACHE_ENABLED) { + return () => { + // noop + }; + } + + const intervalMs = env.CACHE_SWEEP_INTERVAL_MINUTES * 60 * 1000; + const timer = setInterval(() => { + void enforceCacheRetention(); + }, intervalMs); + + timer.unref(); + + return () => { + clearInterval(timer); + }; +} diff --git a/apps/server/src/utils/job-handler.test.ts b/apps/server/src/utils/job-handler.test.ts index 29f52fa..bc2013a 100644 --- a/apps/server/src/utils/job-handler.test.ts +++ b/apps/server/src/utils/job-handler.test.ts @@ -1,30 +1,243 @@ -import { describe, it, expect } from 'vitest'; -import { getOutputFilename } from './job-handler'; +import { describe, it, expect, afterEach, vi } from 'vitest'; +import { writeFile } from 'fs/promises'; +import { clearCacheTestEnv, createTempDirTracker, setCacheTestEnv } from '../test-utils/test-helpers'; + +const { createTempDir, cleanupTempDirs } = createTempDirTracker(); + +function createInputFile(): File { + return new File([Buffer.from('input-media')], 'input.wav', { type: 'audio/wav' }); +} + +async function loadJobHandler(options: { + tempDir: string; + cacheDir: string; + cacheEnabled: boolean; + addJobImpl: ( + jobType: string, + data: Record + ) => Promise<{ waitUntilFinished: () => Promise }>; +}) { + vi.resetModules(); + vi.clearAllMocks(); + + setCacheTestEnv({ + tempDir: options.tempDir, + cacheDir: options.cacheDir, + cacheEnabled: options.cacheEnabled + }); + + const addJobMock = vi.fn(options.addJobImpl); + const validateJobResultMock = vi.fn((result: unknown) => result); + + vi.doMock('~/queue', () => ({ + addJob: addJobMock, + queueEvents: {}, + validateJobResult: validateJobResultMock, + JobTypeName: {} + })); + + const module = await import('./job-handler'); + return { + ...module, + addJobMock, + validateJobResultMock + }; +} + +afterEach(async () => { + await cleanupTempDirs(); + clearCacheTestEnv(); +}); + +describe('processMediaJob cache behavior', () => { + it('should skip queue on cache hit for identical binary requests', async () => { + const tempDir = await createTempDir('job-handler-cache-temp-'); + const cacheDir = await createTempDir('job-handler-cache-dir-'); + + const { processMediaJob, addJobMock } = await loadJobHandler({ + tempDir, + cacheDir, + cacheEnabled: true, + addJobImpl: async (_jobType, data) => { + const outputPath = data['outputPath'] as string; + await writeFile(outputPath, Buffer.from('converted-audio')); + return { + waitUntilFinished: async () => ({ + success: true, + outputPath + }) + }; + } + }); + + const result1 = await processMediaJob({ + file: createInputFile(), + jobType: 'audio:mp3', + outputExtension: 'mp3', + jobData: ({ inputPath, outputPath }) => ({ + inputPath, + outputPath, + quality: 2 + }) + }); + + expect(result1.success).toBe(true); + if (result1.success) { + expect(result1.outputBuffer?.toString()).toBe('converted-audio'); + } + + const result2 = await processMediaJob({ + file: createInputFile(), + jobType: 'audio:mp3', + outputExtension: 'mp3', + jobData: ({ inputPath, outputPath }) => ({ + inputPath, + outputPath, + quality: 2 + }) + }); + + expect(result2.success).toBe(true); + if (result2.success) { + expect(result2.outputBuffer?.toString()).toBe('converted-audio'); + } + + expect(addJobMock).toHaveBeenCalledTimes(1); + }); + + it('should miss cache when processing parameters differ', async () => { + const tempDir = await createTempDir('job-handler-param-temp-'); + const cacheDir = await createTempDir('job-handler-param-cache-'); + + const { processMediaJob, addJobMock } = await loadJobHandler({ + tempDir, + cacheDir, + cacheEnabled: true, + addJobImpl: async (_jobType, data) => { + const outputPath = data['outputPath'] as string; + const quality = data['quality'] as number; + await writeFile(outputPath, Buffer.from(`quality-${quality}`)); + return { + waitUntilFinished: async () => ({ + success: true, + outputPath + }) + }; + } + }); + + const first = await processMediaJob({ + file: createInputFile(), + jobType: 'audio:mp3', + outputExtension: 'mp3', + jobData: ({ inputPath, outputPath }) => ({ + inputPath, + outputPath, + quality: 2 + }) + }); + + const second = await processMediaJob({ + file: createInputFile(), + jobType: 'audio:mp3', + outputExtension: 'mp3', + jobData: ({ inputPath, outputPath }) => ({ + inputPath, + outputPath, + quality: 7 + }) + }); + + expect(first.success).toBe(true); + expect(second.success).toBe(true); + if (first.success && second.success) { + expect(first.outputBuffer?.toString()).toBe('quality-2'); + expect(second.outputBuffer?.toString()).toBe('quality-7'); + } + expect(addJobMock).toHaveBeenCalledTimes(2); + }); + + it('should bypass cache for uploadToS3 jobs', async () => { + const tempDir = await createTempDir('job-handler-s3-temp-'); + const cacheDir = await createTempDir('job-handler-s3-cache-'); + + const { processMediaJob, addJobMock } = await loadJobHandler({ + tempDir, + cacheDir, + cacheEnabled: true, + addJobImpl: async () => { + return { + waitUntilFinished: async () => ({ + success: true, + outputUrl: 'https://example.com/output.mp3' + }) + }; + } + }); + + const first = await processMediaJob({ + file: createInputFile(), + jobType: 'audio:mp3', + outputExtension: 'mp3', + jobData: ({ inputPath, outputPath }) => ({ + inputPath, + outputPath, + quality: 2, + uploadToS3: true + }) + }); + + const second = await processMediaJob({ + file: createInputFile(), + jobType: 'audio:mp3', + outputExtension: 'mp3', + jobData: ({ inputPath, outputPath }) => ({ + inputPath, + outputPath, + quality: 2, + uploadToS3: true + }) + }); + + expect(first.success).toBe(true); + expect(second.success).toBe(true); + if (first.success && second.success) { + expect(first.outputUrl).toBe('https://example.com/output.mp3'); + expect(second.outputUrl).toBe('https://example.com/output.mp3'); + } + expect(addJobMock).toHaveBeenCalledTimes(2); + }); +}); describe('getOutputFilename', () => { - it('should replace extension with new extension', () => { + it('should replace extension with new extension', async () => { + const { getOutputFilename } = await import('./job-handler'); expect(getOutputFilename('video.mp4', 'avi')).toBe('video.avi'); expect(getOutputFilename('audio.wav', 'mp3')).toBe('audio.mp3'); expect(getOutputFilename('image.png', 'jpg')).toBe('image.jpg'); }); - it('should handle files with multiple dots', () => { + it('should handle files with multiple dots', async () => { + const { getOutputFilename } = await import('./job-handler'); expect(getOutputFilename('my.video.file.mp4', 'avi')).toBe('my.video.file.avi'); expect(getOutputFilename('archive.tar.gz', 'zip')).toBe('archive.tar.zip'); }); - it('should return base name without dot when extension is empty', () => { + it('should return base name without dot when extension is empty', async () => { + const { getOutputFilename } = await import('./job-handler'); expect(getOutputFilename('video.mp4', '')).toBe('video'); expect(getOutputFilename('document.pdf', '')).toBe('document'); }); - it('should work correctly for frame extraction filenames', () => { + it('should work correctly for frame extraction filenames', async () => { + const { getOutputFilename } = await import('./job-handler'); const baseName = getOutputFilename('video.mp4', ''); const frameFilename = `${baseName}_frames.zip`; expect(frameFilename).toBe('video_frames.zip'); }); - it('should handle files without extension', () => { + it('should handle files without extension', async () => { + const { getOutputFilename } = await import('./job-handler'); expect(getOutputFilename('README', 'txt')).toBe('README.txt'); expect(getOutputFilename('Makefile', '')).toBe('Makefile'); }); diff --git a/apps/server/src/utils/job-handler.ts b/apps/server/src/utils/job-handler.ts index 9abc92a..b3ffb30 100644 --- a/apps/server/src/utils/job-handler.ts +++ b/apps/server/src/utils/job-handler.ts @@ -4,16 +4,23 @@ import { mkdir, writeFile, readFile, rm } from 'fs/promises'; import path from 'path'; import { env } from '~/config/env'; import { addJob, queueEvents, validateJobResult, JobTypeName } from '~/queue'; - -export const JobPathsSchema = z.object({ +import { + extractCacheableParams, + computeCacheKey, + getCachedOutput, + putCachedOutput, + isCacheEligibleJobData +} from '~/utils/cache'; + +const JobPathsSchema = z.object({ inputPath: z.string(), outputPath: z.string(), jobDir: z.string() }); -export type JobPaths = z.infer; +type JobPaths = z.infer; -export const ProcessJobOptionsSchema = z.object({ +const ProcessJobOptionsSchema = z.object({ file: z.file(), jobType: z.string() as z.ZodType, outputExtension: z.string().min(1), @@ -23,24 +30,20 @@ export const ProcessJobOptionsSchema = z.object({ }) }); -export type ProcessJobOptions = z.infer; - -const SuccessResultSchema = z.object({ - success: z.literal(true), - outputPath: z.string().optional(), - outputUrl: z.string().url().optional(), - outputBuffer: z.instanceof(Buffer).optional(), - metadata: z.record(z.string(), z.unknown()).optional() -}); - -const ErrorResultSchema = z.object({ - success: z.literal(false), - error: z.string() -}); - -export const ProcessJobResultSchema = z.discriminatedUnion('success', [SuccessResultSchema, ErrorResultSchema]); +type ProcessJobOptions = z.infer; -export type ProcessJobResult = z.infer; +type ProcessJobResult = + | { + success: true; + outputPath?: string; + outputUrl?: string; + outputBuffer?: Buffer; + metadata?: Record; + } + | { + success: false; + error: string; + }; export async function processMediaJob(options: ProcessJobOptions): Promise { const validated = ProcessJobOptionsSchema.safeParse(options); @@ -55,22 +58,36 @@ export async function processMediaJob(options: ProcessJobOptions): Promise { await rm(jobDir, { recursive: true, force: true }); }; try { - await mkdir(jobDir, { recursive: true }); - - const inputPath = path.join(jobDir, 'input'); - const outputPath = path.join(jobDir, `output.${outputExtension}`); + const paths: JobPaths = { inputPath, outputPath, jobDir }; + const payload = jobData(paths); + + const inputBuffer = Buffer.from(await file.arrayBuffer()); + const canUseCache = env.CACHE_ENABLED && isCacheEligibleJobData(payload); + const cacheKey = canUseCache ? computeCacheKey(inputBuffer, jobType, extractCacheableParams(payload)) : null; + + if (cacheKey) { + const cached = await getCachedOutput(cacheKey); + if (cached) { + return { + success: true, + outputBuffer: cached.outputBuffer, + metadata: cached.metadata + }; + } + } - const arrayBuffer = await file.arrayBuffer(); - await writeFile(inputPath, Buffer.from(arrayBuffer)); + await mkdir(jobDir, { recursive: true }); + await writeFile(inputPath, inputBuffer); - const paths: JobPaths = { inputPath, outputPath, jobDir }; - const job = await addJob(jobType, jobData(paths)); + const job = await addJob(jobType, payload); const rawResult = await job.waitUntilFinished(queueEvents); const result = validateJobResult(rawResult); @@ -86,6 +103,11 @@ export async function processMediaJob(options: ProcessJobOptions): Promise= 8" } }, + "node_modules/@npmcli/fs": { + "version": "5.0.0", + "resolved": "https://registry.npmjs.org/@npmcli/fs/-/fs-5.0.0.tgz", + "integrity": "sha512-7OsC1gNORBEawOa5+j2pXN9vsicaIOH5cPXxoR6fJOmH6/EXpJB2CajXOu1fPRFun2m1lktEFX11+P89hqO/og==", + "license": "ISC", + "dependencies": { + "semver": "^7.3.5" + }, + "engines": { + "node": "^20.17.0 || >=22.9.0" + } + }, "node_modules/@octokit/auth-token": { "version": "6.0.0", "resolved": "https://registry.npmjs.org/@octokit/auth-token/-/auth-token-6.0.0.tgz", @@ -6849,6 +6865,17 @@ "@types/readdir-glob": "*" } }, + "node_modules/@types/cacache": { + "version": "20.0.1", + "resolved": "https://registry.npmjs.org/@types/cacache/-/cacache-20.0.1.tgz", + "integrity": "sha512-QlKW3AFoFr/hvPHwFHMIVUH/ZCYeetBNou3PCmxu5LaNDvrtBlPJtIA6uhmU9JRt9oxj7IYoqoLcpxtzpPiTcw==", + "dev": true, + "license": "MIT", + "dependencies": { + "@types/node": "*", + "minipass": "*" + } + }, "node_modules/@types/chai": { "version": "5.2.2", "resolved": "https://registry.npmjs.org/@types/chai/-/chai-5.2.2.tgz", @@ -8345,6 +8372,106 @@ "node": ">=8" } }, + "node_modules/cacache": { + "version": "20.0.3", + "resolved": "https://registry.npmjs.org/cacache/-/cacache-20.0.3.tgz", + "integrity": "sha512-3pUp4e8hv07k1QlijZu6Kn7c9+ZpWWk4j3F8N3xPuCExULobqJydKYOTj1FTq58srkJsXvO7LbGAH4C0ZU3WGw==", + "license": "ISC", + "dependencies": { + "@npmcli/fs": "^5.0.0", + "fs-minipass": "^3.0.0", + "glob": "^13.0.0", + "lru-cache": "^11.1.0", + "minipass": "^7.0.3", + "minipass-collect": "^2.0.1", + "minipass-flush": "^1.0.5", + "minipass-pipeline": "^1.2.4", + "p-map": "^7.0.2", + "ssri": "^13.0.0", + "unique-filename": "^5.0.0" + }, + "engines": { + "node": "^20.17.0 || >=22.9.0" + } + }, + "node_modules/cacache/node_modules/balanced-match": { + "version": "4.0.4", + "resolved": "https://registry.npmjs.org/balanced-match/-/balanced-match-4.0.4.tgz", + "integrity": "sha512-BLrgEcRTwX2o6gGxGOCNyMvGSp35YofuYzw9h1IMTRmKqttAZZVU67bdb9Pr2vUHA8+j3i2tJfjO6C6+4myGTA==", + "license": "MIT", + "engines": { + "node": "18 || 20 || >=22" + } + }, + "node_modules/cacache/node_modules/brace-expansion": { + "version": "5.0.4", + "resolved": "https://registry.npmjs.org/brace-expansion/-/brace-expansion-5.0.4.tgz", + "integrity": "sha512-h+DEnpVvxmfVefa4jFbCf5HdH5YMDXRsmKflpf1pILZWRFlTbJpxeU55nJl4Smt5HQaGzg1o6RHFPJaOqnmBDg==", + "license": "MIT", + "dependencies": { + "balanced-match": "^4.0.2" + }, + "engines": { + "node": "18 || 20 || >=22" + } + }, + "node_modules/cacache/node_modules/glob": { + "version": "13.0.6", + "resolved": "https://registry.npmjs.org/glob/-/glob-13.0.6.tgz", + "integrity": "sha512-Wjlyrolmm8uDpm/ogGyXZXb1Z+Ca2B8NbJwqBVg0axK9GbBeoS7yGV6vjXnYdGm6X53iehEuxxbyiKp8QmN4Vw==", + "license": "BlueOak-1.0.0", + "dependencies": { + "minimatch": "^10.2.2", + "minipass": "^7.1.3", + "path-scurry": "^2.0.2" + }, + "engines": { + "node": "18 || 20 || >=22" + }, + "funding": { + "url": "https://github.com/sponsors/isaacs" + } + }, + "node_modules/cacache/node_modules/lru-cache": { + "version": "11.2.6", + "resolved": "https://registry.npmjs.org/lru-cache/-/lru-cache-11.2.6.tgz", + "integrity": "sha512-ESL2CrkS/2wTPfuend7Zhkzo2u0daGJ/A2VucJOgQ/C48S/zB8MMeMHSGKYpXhIjbPxfuezITkaBH1wqv00DDQ==", + "license": "BlueOak-1.0.0", + "engines": { + "node": "20 || >=22" + } + }, + "node_modules/cacache/node_modules/minimatch": { + "version": "10.2.4", + "resolved": "https://registry.npmjs.org/minimatch/-/minimatch-10.2.4.tgz", + "integrity": "sha512-oRjTw/97aTBN0RHbYCdtF1MQfvusSIBQM0IZEgzl6426+8jSC0nF1a/GmnVLpfB9yyr6g6FTqWqiZVbxrtaCIg==", + "license": "BlueOak-1.0.0", + "dependencies": { + "brace-expansion": "^5.0.2" + }, + "engines": { + "node": "18 || 20 || >=22" + }, + "funding": { + "url": "https://github.com/sponsors/isaacs" + } + }, + "node_modules/cacache/node_modules/path-scurry": { + "version": "2.0.2", + "resolved": "https://registry.npmjs.org/path-scurry/-/path-scurry-2.0.2.tgz", + "integrity": "sha512-3O/iVVsJAPsOnpwWIeD+d6z/7PmqApyQePUtCndjatj/9I5LylHvt5qluFaBT3I5h3r1ejfR056c+FCv+NnNXg==", + "license": "BlueOak-1.0.0", + "dependencies": { + "lru-cache": "^11.0.0", + "minipass": "^7.1.2" + }, + "engines": { + "node": "18 || 20 || >=22" + }, + "funding": { + "url": "https://github.com/sponsors/isaacs" + } + }, "node_modules/callsites": { "version": "3.1.0", "resolved": "https://registry.npmjs.org/callsites/-/callsites-3.1.0.tgz", @@ -10830,6 +10957,18 @@ "node": ">=14.14" } }, + "node_modules/fs-minipass": { + "version": "3.0.3", + "resolved": "https://registry.npmjs.org/fs-minipass/-/fs-minipass-3.0.3.tgz", + "integrity": "sha512-XUBA9XClHbnJWSfBzjkm6RvPsyg3sryZt06BEQoXcF7EK/xpGaQYJgQKDJSUH5SGZ76Y7pFx1QBnXz09rU5Fbw==", + "license": "ISC", + "dependencies": { + "minipass": "^7.0.3" + }, + "engines": { + "node": "^14.17.0 || ^16.13.0 || >=18.0.0" + } + }, "node_modules/fsevents": { "version": "2.3.3", "resolved": "https://registry.npmjs.org/fsevents/-/fsevents-2.3.3.tgz", @@ -11726,7 +11865,6 @@ "version": "0.1.4", "resolved": "https://registry.npmjs.org/imurmurhash/-/imurmurhash-0.1.4.tgz", "integrity": "sha512-JmXMZ6wuvDmLiHEml9ykzqO6lwFbof0GG4IkcGaENdCRDDmMVnny7s5HsIgHCbaq0w2MyPhDqkhTUgS2LU2PHA==", - "dev": true, "license": "MIT", "engines": { "node": ">=0.8.19" @@ -13959,6 +14097,78 @@ "node": ">=16 || 14 >=14.17" } }, + "node_modules/minipass-collect": { + "version": "2.0.1", + "resolved": "https://registry.npmjs.org/minipass-collect/-/minipass-collect-2.0.1.tgz", + "integrity": "sha512-D7V8PO9oaz7PWGLbCACuI1qEOsq7UKfLotx/C0Aet43fCUB/wfQ7DYeq2oR/svFJGYDHPr38SHATeaj/ZoKHKw==", + "license": "ISC", + "dependencies": { + "minipass": "^7.0.3" + }, + "engines": { + "node": ">=16 || 14 >=14.17" + } + }, + "node_modules/minipass-flush": { + "version": "1.0.5", + "resolved": "https://registry.npmjs.org/minipass-flush/-/minipass-flush-1.0.5.tgz", + "integrity": "sha512-JmQSYYpPUqX5Jyn1mXaRwOda1uQ8HP5KAT/oDSLCzt1BYRhQU0/hDtsB1ufZfEEzMZ9aAVmsBw8+FWsIXlClWw==", + "license": "ISC", + "dependencies": { + "minipass": "^3.0.0" + }, + "engines": { + "node": ">= 8" + } + }, + "node_modules/minipass-flush/node_modules/minipass": { + "version": "3.3.6", + "resolved": "https://registry.npmjs.org/minipass/-/minipass-3.3.6.tgz", + "integrity": "sha512-DxiNidxSEK+tHG6zOIklvNOwm3hvCrbUrdtzY74U6HKTJxvIDfOUL5W5P2Ghd3DTkhhKPYGqeNUIh5qcM4YBfw==", + "license": "ISC", + "dependencies": { + "yallist": "^4.0.0" + }, + "engines": { + "node": ">=8" + } + }, + "node_modules/minipass-flush/node_modules/yallist": { + "version": "4.0.0", + "resolved": "https://registry.npmjs.org/yallist/-/yallist-4.0.0.tgz", + "integrity": "sha512-3wdGidZyq5PB084XLES5TpOSRA3wjXAlIWMhum2kRcv/41Sn2emQ0dycQW4uZXLejwKvg6EsvbdlVL+FYEct7A==", + "license": "ISC" + }, + "node_modules/minipass-pipeline": { + "version": "1.2.4", + "resolved": "https://registry.npmjs.org/minipass-pipeline/-/minipass-pipeline-1.2.4.tgz", + "integrity": "sha512-xuIq7cIOt09RPRJ19gdi4b+RiNvDFYe5JH+ggNvBqGqpQXcru3PcRmOZuHBKWK1Txf9+cQ+HMVN4d6z46LZP7A==", + "license": "ISC", + "dependencies": { + "minipass": "^3.0.0" + }, + "engines": { + "node": ">=8" + } + }, + "node_modules/minipass-pipeline/node_modules/minipass": { + "version": "3.3.6", + "resolved": "https://registry.npmjs.org/minipass/-/minipass-3.3.6.tgz", + "integrity": "sha512-DxiNidxSEK+tHG6zOIklvNOwm3hvCrbUrdtzY74U6HKTJxvIDfOUL5W5P2Ghd3DTkhhKPYGqeNUIh5qcM4YBfw==", + "license": "ISC", + "dependencies": { + "yallist": "^4.0.0" + }, + "engines": { + "node": ">=8" + } + }, + "node_modules/minipass-pipeline/node_modules/yallist": { + "version": "4.0.0", + "resolved": "https://registry.npmjs.org/yallist/-/yallist-4.0.0.tgz", + "integrity": "sha512-3wdGidZyq5PB084XLES5TpOSRA3wjXAlIWMhum2kRcv/41Sn2emQ0dycQW4uZXLejwKvg6EsvbdlVL+FYEct7A==", + "license": "ISC" + }, "node_modules/minizlib": { "version": "3.1.0", "resolved": "https://registry.npmjs.org/minizlib/-/minizlib-3.1.0.tgz", @@ -16752,7 +16962,6 @@ "version": "7.0.4", "resolved": "https://registry.npmjs.org/p-map/-/p-map-7.0.4.tgz", "integrity": "sha512-tkAQEw8ysMzmkhgw8k+1U/iPhWNhykKnSk4Rd5zLoPJCuJaGRPo6YposrZgaxHKzDHdDWWZvE/Sk7hsL2X/CpQ==", - "dev": true, "license": "MIT", "engines": { "node": ">=18" @@ -18831,6 +19040,18 @@ "nan": "^2.23.0" } }, + "node_modules/ssri": { + "version": "13.0.1", + "resolved": "https://registry.npmjs.org/ssri/-/ssri-13.0.1.tgz", + "integrity": "sha512-QUiRf1+u9wPTL/76GTYlKttDEBWV1ga9ZXW8BG6kfdeyyM8LGPix9gROyg9V2+P0xNyF3X2Go526xKFdMZrHSQ==", + "license": "ISC", + "dependencies": { + "minipass": "^7.0.3" + }, + "engines": { + "node": "^20.17.0 || >=22.9.0" + } + }, "node_modules/stackback": { "version": "0.0.2", "resolved": "https://registry.npmjs.org/stackback/-/stackback-0.0.2.tgz", @@ -19828,6 +20049,30 @@ "url": "https://opencollective.com/unified" } }, + "node_modules/unique-filename": { + "version": "5.0.0", + "resolved": "https://registry.npmjs.org/unique-filename/-/unique-filename-5.0.0.tgz", + "integrity": "sha512-2RaJTAvAb4owyjllTfXzFClJ7WsGxlykkPvCr9pA//LD9goVq+m4PPAeBgNodGZ7nSrntT/auWpJ6Y5IFXcfjg==", + "license": "ISC", + "dependencies": { + "unique-slug": "^6.0.0" + }, + "engines": { + "node": "^20.17.0 || >=22.9.0" + } + }, + "node_modules/unique-slug": { + "version": "6.0.0", + "resolved": "https://registry.npmjs.org/unique-slug/-/unique-slug-6.0.0.tgz", + "integrity": "sha512-4Lup7Ezn8W3d52/xBhZBVdx323ckxa7DEvd9kPQHppTkLoJXw6ltrBCyj5pnrxj0qKDxYMJ56CoxNuFCscdTiw==", + "license": "ISC", + "dependencies": { + "imurmurhash": "^0.1.4" + }, + "engines": { + "node": "^20.17.0 || >=22.9.0" + } + }, "node_modules/unique-string": { "version": "3.0.0", "resolved": "https://registry.npmjs.org/unique-string/-/unique-string-3.0.0.tgz", From 470768dadb1d9d9a1faa99d5e26040598a7b2284 Mon Sep 17 00:00:00 2001 From: crisog Date: Sat, 28 Feb 2026 22:03:43 -0400 Subject: [PATCH 2/7] refactor: simplify cache keying with stable operation signatures --- apps/server/package.json | 1 + apps/server/src/utils/cache.test.ts | 61 +++++++++++++----- apps/server/src/utils/cache.ts | 93 +++++++++------------------- apps/server/src/utils/job-handler.ts | 28 ++++++--- package-lock.json | 1 + 5 files changed, 94 insertions(+), 90 deletions(-) diff --git a/apps/server/package.json b/apps/server/package.json index 508e260..170851b 100644 --- a/apps/server/package.json +++ b/apps/server/package.json @@ -24,6 +24,7 @@ "ioredis": "^5.8.0", "pino": "^10.0.0", "pino-pretty": "^13.1.1", + "safe-stable-stringify": "^2.5.0", "zod": "^4.1.11" }, "devDependencies": { diff --git a/apps/server/src/utils/cache.test.ts b/apps/server/src/utils/cache.test.ts index df1811e..8d04b04 100644 --- a/apps/server/src/utils/cache.test.ts +++ b/apps/server/src/utils/cache.test.ts @@ -40,8 +40,8 @@ describe('cache utility', () => { const input = Buffer.from('same-input'); const params = { quality: 2, mode: 'fit' }; - const key1 = computeCacheKey(input, 'audio:mp3', params); - const key2 = computeCacheKey(input, 'audio:mp3', params); + const key1 = computeCacheKey(input, 'audio:mp3', 'mp3', params); + const key2 = computeCacheKey(input, 'audio:mp3', 'mp3', params); expect(key1).toBe(key2); }); @@ -50,33 +50,64 @@ describe('cache utility', () => { const { computeCacheKey } = await loadCacheModule(); const input = Buffer.from('same-input'); - const keyA = computeCacheKey(input, 'audio:mp3', { quality: 2 }); - const keyB = computeCacheKey(input, 'audio:mp3', { quality: 7 }); - const keyC = computeCacheKey(input, 'video:mp4', { quality: 2 }); + const keyA = computeCacheKey(input, 'audio:mp3', 'mp3', { quality: 2 }); + const keyB = computeCacheKey(input, 'audio:mp3', 'mp3', { quality: 7 }); + const keyC = computeCacheKey(input, 'video:mp4', 'mp4', { quality: 2 }); expect(keyA).not.toBe(keyB); expect(keyA).not.toBe(keyC); }); - it('should strip runtime path and S3 keys from cacheable params', async () => { - const { extractCacheableParams } = await loadCacheModule(); - const params = extractCacheableParams({ - inputPath: '/tmp/input', - outputPath: '/tmp/output', - outputDir: '/tmp/frames', - jobDir: '/tmp/job', + it('should ignore runtime-only path keys in operation signature', async () => { + const { computeCacheKey } = await loadCacheModule(); + const input = Buffer.from('same-input'); + + const keyA = computeCacheKey(input, 'audio:mp3', 'mp3', { + inputPath: '/tmp/input-a', + outputPath: '/tmp/output-a', + outputDir: '/tmp/frames-a', + jobDir: '/tmp/job-a', uploadToS3: true, quality: 2, nested: { - outputPath: '/tmp/nested', + outputPath: '/tmp/nested-a', + mode: 'fit' + } + }); + + const keyB = computeCacheKey(input, 'audio:mp3', 'mp3', { + inputPath: '/tmp/input-b', + outputPath: '/tmp/output-b', + outputDir: '/tmp/frames-b', + jobDir: '/tmp/job-b', + uploadToS3: false, + quality: 2, + nested: { + outputPath: '/tmp/nested-b', mode: 'fit' } }); - expect(params).toEqual({ - nested: { mode: 'fit' }, + expect(keyA).toBe(keyB); + }); + + it('should be deterministic across different object key orders', async () => { + const { computeCacheKey } = await loadCacheModule(); + const input = Buffer.from('same-input'); + + const keyA = computeCacheKey(input, 'audio:mp3', 'mp3', { + quality: 2, + mode: 'fit', + nested: { x: 1, y: 2 } + }); + + const keyB = computeCacheKey(input, 'audio:mp3', 'mp3', { + mode: 'fit', + nested: { y: 2, x: 1 }, quality: 2 }); + + expect(keyA).toBe(keyB); }); it('should round-trip cached output', async () => { diff --git a/apps/server/src/utils/cache.ts b/apps/server/src/utils/cache.ts index 0dbeb91..cdd749e 100644 --- a/apps/server/src/utils/cache.ts +++ b/apps/server/src/utils/cache.ts @@ -1,6 +1,7 @@ import { createHash } from 'crypto'; import { mkdir } from 'fs/promises'; import * as cacache from 'cacache'; +import stringify from 'safe-stable-stringify'; import { z } from 'zod'; import { env } from '~/config/env'; import { logger } from '~/config/logger'; @@ -28,61 +29,30 @@ interface CacheEntryLike { metadata?: unknown; } -function normalizeValue(value: unknown): unknown { - if (value === undefined || typeof value === 'function' || typeof value === 'symbol') { - return undefined; - } - - if (value === null || typeof value === 'string' || typeof value === 'boolean') { - return value; - } - - if (typeof value === 'number') { - return Number.isFinite(value) ? value : null; - } - - if (typeof value === 'bigint') { - return value.toString(); - } - - if (value instanceof Date) { - return value.toISOString(); - } - - if (Buffer.isBuffer(value)) { - return value.toString('base64'); - } - - if (Array.isArray(value)) { - return value.map((item) => { - const normalized = normalizeValue(item); - return normalized === undefined ? null : normalized; - }); - } - - if (typeof value === 'object') { - const record = value as Record; - const normalized: Record = {}; +function computeFileHash(fileBuffer: Buffer): string { + return createHash('sha256').update(fileBuffer).digest('hex'); +} - for (const key of Object.keys(record).sort()) { - if (EXCLUDED_KEYS.has(key)) { - continue; - } +function computeOperationHash(jobType: string, outputExtension: string, params: Record): string { + const operationSignature = stringify({ jobType, outputExtension, params }, (key, value) => { + if (EXCLUDED_KEYS.has(key)) { + return undefined; + } - const normalizedValue = normalizeValue(record[key]); - if (normalizedValue !== undefined) { - normalized[key] = normalizedValue; - } + if (typeof value === 'number' && !Number.isFinite(value)) { + return null; } - return normalized; - } + if (typeof value === 'bigint') { + return value.toString(); + } - return value; -} + return value; + }); -function stableStringify(value: unknown): string { - return JSON.stringify(normalizeValue(value)); + return createHash('sha256') + .update(operationSignature ?? 'null') + .digest('hex'); } function isMissingCacheEntryError(error: unknown): boolean { @@ -137,22 +107,15 @@ export function isCacheEligibleJobData(jobDataResult: Record): return uploadToS3 !== true; } -export function extractCacheableParams(jobDataResult: Record): Record { - const normalized = normalizeValue(jobDataResult); - if (!normalized || typeof normalized !== 'object' || Array.isArray(normalized)) { - return {}; - } - return normalized as Record; -} - -export function computeCacheKey(fileBuffer: Buffer, jobType: string, params: Record): string { - const hash = createHash('sha256'); - hash.update(fileBuffer); - hash.update('\0'); - hash.update(jobType); - hash.update('\0'); - hash.update(stableStringify(params)); - return hash.digest('hex'); +export function computeCacheKey( + fileBuffer: Buffer, + jobType: string, + outputExtension: string, + params: Record +): string { + const fileHash = computeFileHash(fileBuffer); + const operationHash = computeOperationHash(jobType, outputExtension, params); + return `${fileHash}:${operationHash}`; } export async function getCachedOutput(cacheKey: string): Promise { diff --git a/apps/server/src/utils/job-handler.ts b/apps/server/src/utils/job-handler.ts index b3ffb30..ad32658 100644 --- a/apps/server/src/utils/job-handler.ts +++ b/apps/server/src/utils/job-handler.ts @@ -4,13 +4,7 @@ import { mkdir, writeFile, readFile, rm } from 'fs/promises'; import path from 'path'; import { env } from '~/config/env'; import { addJob, queueEvents, validateJobResult, JobTypeName } from '~/queue'; -import { - extractCacheableParams, - computeCacheKey, - getCachedOutput, - putCachedOutput, - isCacheEligibleJobData -} from '~/utils/cache'; +import { computeCacheKey, getCachedOutput, putCachedOutput, isCacheEligibleJobData } from '~/utils/cache'; const JobPathsSchema = z.object({ inputPath: z.string(), @@ -71,7 +65,10 @@ export async function processMediaJob(options: ProcessJobOptions): Promise Date: Sat, 28 Feb 2026 22:11:51 -0400 Subject: [PATCH 3/7] test: enhance cache utility and job handler tests for key generation and cache behavior --- apps/server/src/utils/cache.test.ts | 21 ++++ apps/server/src/utils/job-handler.test.ts | 117 +++++++++++++++++++++- 2 files changed, 134 insertions(+), 4 deletions(-) diff --git a/apps/server/src/utils/cache.test.ts b/apps/server/src/utils/cache.test.ts index 8d04b04..650ca56 100644 --- a/apps/server/src/utils/cache.test.ts +++ b/apps/server/src/utils/cache.test.ts @@ -58,6 +58,27 @@ describe('cache utility', () => { expect(keyA).not.toBe(keyC); }); + it('should generate different keys for different input bytes', async () => { + const { computeCacheKey } = await loadCacheModule(); + const params = { quality: 2, mode: 'fit' }; + + const keyA = computeCacheKey(Buffer.from('input-a'), 'audio:mp3', 'mp3', params); + const keyB = computeCacheKey(Buffer.from('input-b'), 'audio:mp3', 'mp3', params); + + expect(keyA).not.toBe(keyB); + }); + + it('should generate different keys for different output extensions', async () => { + const { computeCacheKey } = await loadCacheModule(); + const input = Buffer.from('same-input'); + const params = { quality: 2, mode: 'fit' }; + + const keyA = computeCacheKey(input, 'audio:mp3', 'mp3', params); + const keyB = computeCacheKey(input, 'audio:mp3', 'wav', params); + + expect(keyA).not.toBe(keyB); + }); + it('should ignore runtime-only path keys in operation signature', async () => { const { computeCacheKey } = await loadCacheModule(); const input = Buffer.from('same-input'); diff --git a/apps/server/src/utils/job-handler.test.ts b/apps/server/src/utils/job-handler.test.ts index bc2013a..e139433 100644 --- a/apps/server/src/utils/job-handler.test.ts +++ b/apps/server/src/utils/job-handler.test.ts @@ -1,11 +1,11 @@ import { describe, it, expect, afterEach, vi } from 'vitest'; -import { writeFile } from 'fs/promises'; +import { readFile, writeFile } from 'fs/promises'; import { clearCacheTestEnv, createTempDirTracker, setCacheTestEnv } from '../test-utils/test-helpers'; const { createTempDir, cleanupTempDirs } = createTempDirTracker(); -function createInputFile(): File { - return new File([Buffer.from('input-media')], 'input.wav', { type: 'audio/wav' }); +function createInputFile(content = 'input-media'): File { + return new File([Buffer.from(content)], 'input.wav', { type: 'audio/wav' }); } async function loadJobHandler(options: { @@ -64,7 +64,8 @@ describe('processMediaJob cache behavior', () => { return { waitUntilFinished: async () => ({ success: true, - outputPath + outputPath, + metadata: { codec: 'mp3' } }) }; } @@ -84,6 +85,7 @@ describe('processMediaJob cache behavior', () => { expect(result1.success).toBe(true); if (result1.success) { expect(result1.outputBuffer?.toString()).toBe('converted-audio'); + expect(result1.metadata).toEqual({ codec: 'mp3' }); } const result2 = await processMediaJob({ @@ -100,6 +102,7 @@ describe('processMediaJob cache behavior', () => { expect(result2.success).toBe(true); if (result2.success) { expect(result2.outputBuffer?.toString()).toBe('converted-audio'); + expect(result2.metadata).toEqual({ codec: 'mp3' }); } expect(addJobMock).toHaveBeenCalledTimes(1); @@ -157,6 +160,112 @@ describe('processMediaJob cache behavior', () => { expect(addJobMock).toHaveBeenCalledTimes(2); }); + it('should miss cache when input bytes differ', async () => { + const tempDir = await createTempDir('job-handler-input-temp-'); + const cacheDir = await createTempDir('job-handler-input-cache-'); + + const { processMediaJob, addJobMock } = await loadJobHandler({ + tempDir, + cacheDir, + cacheEnabled: true, + addJobImpl: async (_jobType, data) => { + const inputPath = data['inputPath'] as string; + const outputPath = data['outputPath'] as string; + const inputContent = await readFile(inputPath, 'utf8'); + await writeFile(outputPath, Buffer.from(`from-${inputContent}`)); + return { + waitUntilFinished: async () => ({ + success: true, + outputPath + }) + }; + } + }); + + const first = await processMediaJob({ + file: createInputFile('input-a'), + jobType: 'audio:mp3', + outputExtension: 'mp3', + jobData: ({ inputPath, outputPath }) => ({ + inputPath, + outputPath, + quality: 2 + }) + }); + + const second = await processMediaJob({ + file: createInputFile('input-b'), + jobType: 'audio:mp3', + outputExtension: 'mp3', + jobData: ({ inputPath, outputPath }) => ({ + inputPath, + outputPath, + quality: 2 + }) + }); + + expect(first.success).toBe(true); + expect(second.success).toBe(true); + if (first.success && second.success) { + expect(first.outputBuffer?.toString()).toBe('from-input-a'); + expect(second.outputBuffer?.toString()).toBe('from-input-b'); + } + expect(addJobMock).toHaveBeenCalledTimes(2); + }); + + it('should not reuse cached output when cache is disabled', async () => { + const tempDir = await createTempDir('job-handler-disabled-temp-'); + const cacheDir = await createTempDir('job-handler-disabled-cache-'); + let callCount = 0; + + const { processMediaJob, addJobMock } = await loadJobHandler({ + tempDir, + cacheDir, + cacheEnabled: false, + addJobImpl: async (_jobType, data) => { + callCount += 1; + const outputPath = data['outputPath'] as string; + await writeFile(outputPath, Buffer.from(`call-${callCount}`)); + return { + waitUntilFinished: async () => ({ + success: true, + outputPath + }) + }; + } + }); + + const first = await processMediaJob({ + file: createInputFile(), + jobType: 'audio:mp3', + outputExtension: 'mp3', + jobData: ({ inputPath, outputPath }) => ({ + inputPath, + outputPath, + quality: 2 + }) + }); + + const second = await processMediaJob({ + file: createInputFile(), + jobType: 'audio:mp3', + outputExtension: 'mp3', + jobData: ({ inputPath, outputPath }) => ({ + inputPath, + outputPath, + quality: 2 + }) + }); + + expect(first.success).toBe(true); + expect(second.success).toBe(true); + if (first.success && second.success) { + expect(first.outputBuffer?.toString()).toBe('call-1'); + expect(second.outputBuffer?.toString()).toBe('call-2'); + } + expect(addJobMock).toHaveBeenCalledTimes(2); + }); + it('should bypass cache for uploadToS3 jobs', async () => { const tempDir = await createTempDir('job-handler-s3-temp-'); const cacheDir = await createTempDir('job-handler-s3-cache-'); From f4e853a6471de34a88a6a8f99dd2e9b16ecffcde Mon Sep 17 00:00:00 2001 From: crisog Date: Sat, 28 Feb 2026 22:20:44 -0400 Subject: [PATCH 4/7] refactor: replace custom boolean schema with z.stringbool for CACHE_ENABLED --- apps/server/src/config/env.ts | 18 +----------------- 1 file changed, 1 insertion(+), 17 deletions(-) diff --git a/apps/server/src/config/env.ts b/apps/server/src/config/env.ts index 8482aa8..de65de9 100644 --- a/apps/server/src/config/env.ts +++ b/apps/server/src/config/env.ts @@ -6,22 +6,6 @@ if (process.env['NODE_ENV'] !== 'production') { dotenv.config(); } -const EnvBooleanSchema = z.preprocess((value) => { - if (typeof value !== 'string') { - return value; - } - - const normalized = value.trim().toLowerCase(); - if (['1', 'true', 'yes', 'on'].includes(normalized)) { - return true; - } - if (['0', 'false', 'no', 'off', ''].includes(normalized)) { - return false; - } - - return value; -}, z.boolean()); - const schema = z.object({ PORT: z.coerce.number().default(3000), NODE_ENV: z.enum(['development', 'production', 'test']).default('development'), @@ -32,7 +16,7 @@ const schema = z.object({ MAX_FILE_SIZE: z.coerce.number().default(100 * 1024 * 1024), STORAGE_MODE: z.enum(['stateless', 's3']).default('stateless'), - CACHE_ENABLED: EnvBooleanSchema.default(false), + CACHE_ENABLED: z.stringbool().default(false), CACHE_DIR: z.string().optional(), CACHE_TTL_HOURS: z.coerce.number().int().positive().default(2160), CACHE_MAX_SIZE_MB: z.coerce.number().int().positive().default(1024), From fa17d661ad870f3962e4b76b2b0b8fc3989ad404 Mon Sep 17 00:00:00 2001 From: crisog Date: Sat, 28 Feb 2026 22:42:46 -0400 Subject: [PATCH 5/7] refactor: remove CACHE_SWEEP_INTERVAL_MINUTES --- .env.example | 1 - README.md | 3 +-- apps/server/src/config/env.ts | 1 - apps/server/src/server.ts | 13 +------------ apps/server/src/test-utils/test-helpers.ts | 3 --- apps/server/src/utils/cache.test.ts | 10 ++-------- apps/server/src/utils/cache.ts | 19 ------------------- 7 files changed, 4 insertions(+), 46 deletions(-) diff --git a/.env.example b/.env.example index 0154385..66ff0ff 100644 --- a/.env.example +++ b/.env.example @@ -18,7 +18,6 @@ CACHE_ENABLED=false CACHE_DIR=/tmp/ffmpeg-rest/cache CACHE_TTL_HOURS=2160 CACHE_MAX_SIZE_MB=1024 -CACHE_SWEEP_INTERVAL_MINUTES=60 S3_ENDPOINT=https://s3.amazonaws.com S3_REGION=us-east-1 diff --git a/README.md b/README.md index e902ec7..51ec8dc 100644 --- a/README.md +++ b/README.md @@ -64,7 +64,7 @@ Stateless mode can optionally cache binary conversion outputs using `cacache` to - Cache scope: binary conversion endpoints only (not `/.../url` S3 responses, not `/media/info`) - Cache key: SHA-256 of input bytes + job type + normalized processing params -- Retention: TTL + size cap with periodic cleanup +- Retention: TTL + size cap (enforced on reads/writes and startup) - Storage: local filesystem (ephemeral by default) **Configuration**: @@ -74,7 +74,6 @@ CACHE_ENABLED=false # Enable/disable stateless cache CACHE_DIR=/tmp/ffmpeg-rest/cache CACHE_TTL_HOURS=2160 # 90 days CACHE_MAX_SIZE_MB=1024 # 1 GiB -CACHE_SWEEP_INTERVAL_MINUTES=60 ``` ### S3 Mode diff --git a/apps/server/src/config/env.ts b/apps/server/src/config/env.ts index de65de9..cf69a24 100644 --- a/apps/server/src/config/env.ts +++ b/apps/server/src/config/env.ts @@ -20,7 +20,6 @@ const schema = z.object({ CACHE_DIR: z.string().optional(), CACHE_TTL_HOURS: z.coerce.number().int().positive().default(2160), CACHE_MAX_SIZE_MB: z.coerce.number().int().positive().default(1024), - CACHE_SWEEP_INTERVAL_MINUTES: z.coerce.number().int().positive().default(60), AUTH_TOKEN: z.string().optional() }); diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index 254a815..ac0422d 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -3,25 +3,14 @@ import { createApp } from '~/app'; import { env } from '~/config/env'; import { checkRedisHealth } from '~/config/redis'; import { logger } from '~/config/logger'; -import { initCacheDir, startCacheCleanup } from '~/utils/cache'; +import { initCacheDir } from '~/utils/cache'; await checkRedisHealth(); -let stopCacheCleanup: () => void = () => undefined; if (env.CACHE_ENABLED) { await initCacheDir(); - stopCacheCleanup = startCacheCleanup(); } -const handleShutdown = (signal: NodeJS.Signals) => { - logger.info({ signal }, 'Shutting down server'); - stopCacheCleanup(); - process.exit(0); -}; - -process.once('SIGINT', () => handleShutdown('SIGINT')); -process.once('SIGTERM', () => handleShutdown('SIGTERM')); - const app = createApp(); serve( diff --git a/apps/server/src/test-utils/test-helpers.ts b/apps/server/src/test-utils/test-helpers.ts index 368b2b5..b734d07 100644 --- a/apps/server/src/test-utils/test-helpers.ts +++ b/apps/server/src/test-utils/test-helpers.ts @@ -8,7 +8,6 @@ interface CacheTestEnvOptions { cacheEnabled?: boolean; ttlHours?: number; maxSizeMb?: number; - sweepIntervalMinutes?: number; } export function createTempDirTracker() { @@ -40,7 +39,6 @@ export function setCacheTestEnv(options: CacheTestEnvOptions): void { process.env['CACHE_DIR'] = options.cacheDir; process.env['CACHE_TTL_HOURS'] = String(options.ttlHours ?? 24); process.env['CACHE_MAX_SIZE_MB'] = String(options.maxSizeMb ?? 10); - process.env['CACHE_SWEEP_INTERVAL_MINUTES'] = String(options.sweepIntervalMinutes ?? 60); } export function clearCacheTestEnv(): void { @@ -48,6 +46,5 @@ export function clearCacheTestEnv(): void { delete process.env['CACHE_DIR']; delete process.env['CACHE_TTL_HOURS']; delete process.env['CACHE_MAX_SIZE_MB']; - delete process.env['CACHE_SWEEP_INTERVAL_MINUTES']; delete process.env['TEMP_DIR']; } diff --git a/apps/server/src/utils/cache.test.ts b/apps/server/src/utils/cache.test.ts index 650ca56..57a2e6c 100644 --- a/apps/server/src/utils/cache.test.ts +++ b/apps/server/src/utils/cache.test.ts @@ -4,12 +4,7 @@ import { clearCacheTestEnv, createTempDirTracker, setCacheTestEnv } from '../tes const { createTempDir, cleanupTempDirs } = createTempDirTracker(); -async function loadCacheModule(options?: { - cacheEnabled?: boolean; - ttlHours?: number; - maxSizeMb?: number; - sweepIntervalMinutes?: number; -}) { +async function loadCacheModule(options?: { cacheEnabled?: boolean; ttlHours?: number; maxSizeMb?: number }) { const cacheDir = await createTempDir('cache-utils-'); const tempDir = await createTempDir('cache-utils-temp-'); @@ -21,8 +16,7 @@ async function loadCacheModule(options?: { cacheDir, cacheEnabled: options?.cacheEnabled, ttlHours: options?.ttlHours, - maxSizeMb: options?.maxSizeMb, - sweepIntervalMinutes: options?.sweepIntervalMinutes + maxSizeMb: options?.maxSizeMb }); const mod = await import('./cache'); diff --git a/apps/server/src/utils/cache.ts b/apps/server/src/utils/cache.ts index cdd749e..d131bfa 100644 --- a/apps/server/src/utils/cache.ts +++ b/apps/server/src/utils/cache.ts @@ -239,22 +239,3 @@ export async function initCacheDir(): Promise { await mkdir(env.CACHE_DIR, { recursive: true }); await enforceCacheRetention(); } - -export function startCacheCleanup(): () => void { - if (!env.CACHE_ENABLED) { - return () => { - // noop - }; - } - - const intervalMs = env.CACHE_SWEEP_INTERVAL_MINUTES * 60 * 1000; - const timer = setInterval(() => { - void enforceCacheRetention(); - }, intervalMs); - - timer.unref(); - - return () => { - clearInterval(timer); - }; -} From 829f4e4558c2511eb9fb116aa37001c2534db841 Mon Sep 17 00:00:00 2001 From: crisog Date: Sat, 28 Feb 2026 22:50:37 -0400 Subject: [PATCH 6/7] refactor: simplify putCachedOutput function by removing unused parameters --- apps/server/src/utils/cache.test.ts | 8 ++++---- apps/server/src/utils/cache.ts | 23 ----------------------- apps/server/src/utils/job-handler.ts | 9 +++------ 3 files changed, 7 insertions(+), 33 deletions(-) diff --git a/apps/server/src/utils/cache.test.ts b/apps/server/src/utils/cache.test.ts index 57a2e6c..ee02f96 100644 --- a/apps/server/src/utils/cache.test.ts +++ b/apps/server/src/utils/cache.test.ts @@ -131,7 +131,7 @@ describe('cache utility', () => { const key = 'roundtrip-key'; const output = Buffer.from('converted-output'); - await putCachedOutput(key, output, 'audio:mp3', 'mp3', { codec: 'mp3' }); + await putCachedOutput(key, output, { codec: 'mp3' }); const cached = await getCachedOutput(key); expect(cached?.outputBuffer.toString()).toBe('converted-output'); @@ -163,9 +163,9 @@ describe('cache utility', () => { const keyNew = 'new-entry'; const payload = Buffer.alloc(700 * 1024, 1); - await putCachedOutput(keyOld, payload, 'audio:mp3', 'mp3'); + await putCachedOutput(keyOld, payload); await new Promise((resolve) => setTimeout(resolve, 10)); - await putCachedOutput(keyNew, payload, 'audio:mp3', 'mp3'); + await putCachedOutput(keyNew, payload); const oldEntry = await getCachedOutput(keyOld); const newEntry = await getCachedOutput(keyNew); @@ -176,7 +176,7 @@ describe('cache utility', () => { it('should no-op when cache is disabled', async () => { const { putCachedOutput, getCachedOutput } = await loadCacheModule({ cacheEnabled: false }); - await putCachedOutput('disabled-key', Buffer.from('data'), 'audio:mp3', 'mp3'); + await putCachedOutput('disabled-key', Buffer.from('data')); const cached = await getCachedOutput('disabled-key'); expect(cached).toBeNull(); diff --git a/apps/server/src/utils/cache.ts b/apps/server/src/utils/cache.ts index d131bfa..ff93435 100644 --- a/apps/server/src/utils/cache.ts +++ b/apps/server/src/utils/cache.ts @@ -10,8 +10,6 @@ const EXCLUDED_KEYS = new Set(['inputPath', 'outputPath', 'outputDir', 'jobDir', const CacheMetadataSchema = z.object({ createdAt: z.number(), - jobType: z.string(), - outputExtension: z.string(), resultMetadata: z.record(z.string(), z.unknown()).optional() }); @@ -90,18 +88,6 @@ function isExpiredEntry(entry: CacheEntryLike): boolean { return isExpiredByTimestamp(getEntryTimestamp(entry)); } -async function runVerifyAfterRemovals(hadRemovals: boolean): Promise { - if (!hadRemovals) { - return; - } - - try { - await cacache.verify(env.CACHE_DIR, { concurrency: 8 }); - } catch (error) { - logger.warn({ error }, 'Cache verify failed after retention cleanup'); - } -} - export function isCacheEligibleJobData(jobDataResult: Record): boolean { const uploadToS3 = jobDataResult['uploadToS3']; return uploadToS3 !== true; @@ -165,14 +151,12 @@ async function enforceCacheRetention(requiredSpaceBytes = 0): Promise { try { const entriesMap = await cacache.ls(env.CACHE_DIR); const entries = Object.values(entriesMap) as CacheEntryLike[]; - let hadRemovals = false; const activeEntries: CacheEntryLike[] = []; for (const entry of entries) { if (isExpiredEntry(entry)) { await cacache.rm.entry(env.CACHE_DIR, entry.key); - hadRemovals = true; } else { activeEntries.push(entry); } @@ -189,10 +173,7 @@ async function enforceCacheRetention(requiredSpaceBytes = 0): Promise { await cacache.rm.entry(env.CACHE_DIR, oldest.key); currentSize -= oldest.size; - hadRemovals = true; } - - await runVerifyAfterRemovals(hadRemovals); } catch (error) { logger.warn({ error }, 'Failed to enforce cache retention'); } @@ -201,8 +182,6 @@ async function enforceCacheRetention(requiredSpaceBytes = 0): Promise { export async function putCachedOutput( cacheKey: string, outputBuffer: Buffer, - jobType: string, - outputExtension: string, metadata?: Record ): Promise { if (!env.CACHE_ENABLED) { @@ -221,8 +200,6 @@ export async function putCachedOutput( await cacache.put(env.CACHE_DIR, cacheKey, outputBuffer, { metadata: { createdAt: Date.now(), - jobType, - outputExtension, resultMetadata: metadata } }); diff --git a/apps/server/src/utils/job-handler.ts b/apps/server/src/utils/job-handler.ts index ad32658..eacc082 100644 --- a/apps/server/src/utils/job-handler.ts +++ b/apps/server/src/utils/job-handler.ts @@ -89,12 +89,10 @@ export async function processMediaJob(options: ProcessJobOptions): Promise Date: Sat, 28 Feb 2026 23:12:09 -0400 Subject: [PATCH 7/7] test: add garbage collection test for unreferenced content in cache utility --- apps/server/src/utils/cache.test.ts | 44 +++++++++++++++++++++++++++++ apps/server/src/utils/cache.ts | 19 +++++++++++++ 2 files changed, 63 insertions(+) diff --git a/apps/server/src/utils/cache.test.ts b/apps/server/src/utils/cache.test.ts index ee02f96..707cf2c 100644 --- a/apps/server/src/utils/cache.test.ts +++ b/apps/server/src/utils/cache.test.ts @@ -1,9 +1,41 @@ import { describe, it, expect, afterEach, vi } from 'vitest'; import * as cacache from 'cacache'; +import { readdir } from 'fs/promises'; +import path from 'path'; import { clearCacheTestEnv, createTempDirTracker, setCacheTestEnv } from '../test-utils/test-helpers'; const { createTempDir, cleanupTempDirs } = createTempDirTracker(); +async function countContentBlobs(cacheDir: string): Promise { + const contentRoot = path.join(cacheDir, 'content-v2'); + let count = 0; + + const walk = async (dir: string): Promise => { + const entries = await readdir(dir, { withFileTypes: true }); + + for (const entry of entries) { + const fullPath = path.join(dir, entry.name); + if (entry.isDirectory()) { + await walk(fullPath); + } else if (entry.isFile()) { + count += 1; + } + } + }; + + try { + await walk(contentRoot); + return count; + } catch (error) { + const maybeErrno = error as NodeJS.ErrnoException; + if (maybeErrno.code === 'ENOENT') { + return 0; + } + + throw error; + } +} + async function loadCacheModule(options?: { cacheEnabled?: boolean; ttlHours?: number; maxSizeMb?: number }) { const cacheDir = await createTempDir('cache-utils-'); const tempDir = await createTempDir('cache-utils-temp-'); @@ -174,6 +206,18 @@ describe('cache utility', () => { expect(newEntry).not.toBeNull(); }); + it('should garbage collect unreferenced content after eviction', async () => { + const { initCacheDir, putCachedOutput, cacheDir } = await loadCacheModule({ maxSizeMb: 1 }); + await initCacheDir(); + + await putCachedOutput('entry-a', Buffer.alloc(700 * 1024, 1)); + await new Promise((resolve) => setTimeout(resolve, 10)); + await putCachedOutput('entry-b', Buffer.alloc(700 * 1024, 2)); + + const contentBlobs = await countContentBlobs(cacheDir); + expect(contentBlobs).toBe(1); + }); + it('should no-op when cache is disabled', async () => { const { putCachedOutput, getCachedOutput } = await loadCacheModule({ cacheEnabled: false }); await putCachedOutput('disabled-key', Buffer.from('data')); diff --git a/apps/server/src/utils/cache.ts b/apps/server/src/utils/cache.ts index ff93435..2108a93 100644 --- a/apps/server/src/utils/cache.ts +++ b/apps/server/src/utils/cache.ts @@ -88,6 +88,18 @@ function isExpiredEntry(entry: CacheEntryLike): boolean { return isExpiredByTimestamp(getEntryTimestamp(entry)); } +async function runVerifyAfterRemovals(hadRemovals: boolean): Promise { + if (!hadRemovals) { + return; + } + + try { + await cacache.verify(env.CACHE_DIR, { concurrency: 8 }); + } catch (error) { + logger.warn({ error }, 'Cache verify failed after retention cleanup'); + } +} + export function isCacheEligibleJobData(jobDataResult: Record): boolean { const uploadToS3 = jobDataResult['uploadToS3']; return uploadToS3 !== true; @@ -115,12 +127,14 @@ export async function getCachedOutput(cacheKey: string): Promise { try { const entriesMap = await cacache.ls(env.CACHE_DIR); const entries = Object.values(entriesMap) as CacheEntryLike[]; + let hadRemovals = false; const activeEntries: CacheEntryLike[] = []; for (const entry of entries) { if (isExpiredEntry(entry)) { await cacache.rm.entry(env.CACHE_DIR, entry.key); + hadRemovals = true; } else { activeEntries.push(entry); } @@ -173,7 +189,10 @@ async function enforceCacheRetention(requiredSpaceBytes = 0): Promise { await cacache.rm.entry(env.CACHE_DIR, oldest.key); currentSize -= oldest.size; + hadRemovals = true; } + + await runVerifyAfterRemovals(hadRemovals); } catch (error) { logger.warn({ error }, 'Failed to enforce cache retention'); }