refactor(api): drop Bunny for pluggable cache purge adapters (#2742)

fluxer_api/src/api/worker/WorkerDependencies.ts

@@ -17,7 +17,7 @@ import type {GuildAuditLogService} from '@app/api/guild/GuildAuditLogService';
 import type {GuildRepository} from '@app/api/guild/repositories/GuildRepository';
 import type {GuildService} from '@app/api/guild/services/GuildService';
 import type {AvatarService} from '@app/api/infrastructure/AvatarService';
-import type {IPurgeQueue} from '@app/api/infrastructure/BunnyPurgeQueue';
+import type {IPurgeQueue} from '@app/api/infrastructure/CachePurgeQueue';
 import type {DiscriminatorService} from '@app/api/infrastructure/DiscriminatorService';
 import type {EmbedService} from '@app/api/infrastructure/EmbedService';
 import type {IAssetDeletionQueue} from '@app/api/infrastructure/IAssetDeletionQueue';

fluxer_api/src/api/worker/WorkerLaneConfig.ts

@@ -70,7 +70,7 @@ const LANE_CONFIG = {
 			'indexChannelMessages',
 			'indexGuildMembers',
 			'processAssetDeletionQueue',
-			'processBunnyPurgeQueue',
+			'processCachePurgeQueue',
 			'processExpiredPremiumSweep',
 			'processInactivityDeletions',
 			'processPendingBulkMessageDeletions',

fluxer_api/src/api/worker/WorkerMain.ts

@@ -54,8 +54,8 @@ const SEARCH_REQUIRED_TASKS = new Set<string>([
 
 function registerCronJobs(cron: CronScheduler): void {
 	cron.upsert('processAssetDeletionQueue', 'processAssetDeletionQueue', {}, '0 */5 * * * *', {ledger: false});
-	if (Config.bunny.purgeEnabled) {
-		cron.upsert('processBunnyPurgeQueue', 'processBunnyPurgeQueue', {}, '*/10 * * * * *', {ledger: false});
+	if (Config.cachePurge.adapter !== 'none') {
+		cron.upsert('processCachePurgeQueue', 'processCachePurgeQueue', {}, '*/10 * * * * *', {ledger: false});
 	}
 	cron.upsert('processPendingBulkMessageDeletions', 'processPendingBulkMessageDeletions', {}, '0 */10 * * * *', {
 		ledger: false,
@@ -80,7 +80,7 @@ function registerCronJobs(cron: CronScheduler): void {
 	Logger.info(
 		{
 			blocklistFeeds: Config.blocklistFeeds.enabled,
-			bunnyPurge: Config.bunny.purgeEnabled,
+			cachePurgeAdapter: Config.cachePurge.adapter,
 			selfHosted: Config.instance.selfHosted,
 		},
 		'Cron jobs registered successfully',

fluxer_api/src/api/worker/WorkerTaskRegistry.ts

@@ -25,7 +25,7 @@ import indexChannelMessages from '@app/api/worker/tasks/IndexChannelMessages';
 import indexGuildMembers from '@app/api/worker/tasks/IndexGuildMembers';
 import messageShred from '@app/api/worker/tasks/MessageShred';
 import processAssetDeletionQueue from '@app/api/worker/tasks/ProcessAssetDeletionQueue';
-import processBunnyPurgeQueue from '@app/api/worker/tasks/ProcessBunnyPurgeQueue';
+import processCachePurgeQueue from '@app/api/worker/tasks/ProcessCachePurgeQueue';
 import processExpiredPremiumSweep from '@app/api/worker/tasks/ProcessExpiredPremiumSweep';
 import processInactivityDeletions from '@app/api/worker/tasks/ProcessInactivityDeletions';
 import processPendingBulkMessageDeletions from '@app/api/worker/tasks/ProcessPendingBulkMessageDeletions';
@@ -69,7 +69,7 @@ export const workerTasks: Record<WorkerTaskName, WorkerTaskHandler> = {
 	indexGuildMembers,
 	messageShred,
 	processAssetDeletionQueue,
-	processBunnyPurgeQueue,
+	processCachePurgeQueue,
 	processStripeWebhook,
 	processExpiredPremiumSweep,
 	processInactivityDeletions,

fluxer_api/src/api/worker/tasks/ProcessAssetDeletionQueue.ts

@@ -3,7 +3,7 @@
 import {createGuildID, createUserID} from '@app/api/BrandedTypes';
 import {Config} from '@app/api/Config';
 import type {GuildRepository} from '@app/api/guild/repositories/GuildRepository';
-import type {IPurgeQueue} from '@app/api/infrastructure/BunnyPurgeQueue';
+import type {IPurgeQueue} from '@app/api/infrastructure/CachePurgeQueue';
 import type {
 	IAssetDeletionQueue,
 	QueuedAssetDeletion,

fluxer_api/src/api/worker/tasks/ProcessBunnyPurgeQueue.ts

@@ -1,152 +0,0 @@
-// SPDX-License-Identifier: AGPL-3.0-or-later
-
-import {Config} from '@app/api/Config';
-import type {BunnyPurgeQueue} from '@app/api/infrastructure/BunnyPurgeQueue';
-import {Logger} from '@app/api/Logger';
-import {EXTERNAL_RESPONSE_LIMITS} from '@app/api/utils/ExternalResponseLimits';
-import * as FetchUtils from '@app/api/utils/FetchUtils';
-import {getWorkerDependencies} from '@app/api/worker/WorkerContext';
-import {formatUrlForDiagnostics} from '@pkgs/http_client/src/HttpClientDiagnostics';
-import type {WorkerTaskHandler} from '@pkgs/worker/src/contracts/WorkerTask';
-import {ms} from 'itty-time';
-
-const EXACT_BATCH_SIZE = 120;
-const PREFIX_BATCH_SIZE = 20;
-
-type PurgeLabel = 'exact' | 'prefix';
-
-interface PurgeStats {
-	attempted: number;
-	purged: number;
-	requeued: number;
-	failures: number;
-	rateLimited: number;
-}
-
-function createEmptyPurgeStats(): PurgeStats {
-	return {
-		attempted: 0,
-		purged: 0,
-		requeued: 0,
-		failures: 0,
-		rateLimited: 0,
-	};
-}
-
-function mergePurgeStats(target: PurgeStats, source: PurgeStats): void {
-	target.attempted += source.attempted;
-	target.purged += source.purged;
-	target.requeued += source.requeued;
-	target.failures += source.failures;
-	target.rateLimited += source.rateLimited;
-}
-
-async function purgeUrls(
-	urls: Array<string>,
-	apiKey: string,
-	queue: BunnyPurgeQueue,
-	label: PurgeLabel,
-): Promise<PurgeStats> {
-	const stats = createEmptyPurgeStats();
-	for (let index = 0; index < urls.length; index++) {
-		const url = urls[index]!;
-		stats.attempted++;
-		try {
-			const response = await fetch(`https://api.bunny.net/purge?url=${encodeURIComponent(url)}&async=true`, {
-				method: 'POST',
-				headers: {
-					AccessKey: apiKey,
-				},
-				signal: AbortSignal.timeout(ms('30 seconds')),
-			});
-			if (response.status === 429) {
-				FetchUtils.discardResponseBody(response.body, response.status);
-				const remaining = urls.slice(index);
-				await queue.addUrls(remaining);
-				stats.requeued += remaining.length;
-				stats.rateLimited += remaining.length;
-				Logger.warn({label, rateLimitedUrls: remaining.length}, 'Rate limited by Bunny CDN, re-queued remaining URLs');
-				break;
-			}
-			if (!response.ok) {
-				stats.failures++;
-				const errorBody = await FetchUtils.streamToBufferWithLimit(response.body, {
-					maxBytes: EXTERNAL_RESPONSE_LIMITS.bunnyErrorBytes,
-					headers: response.headers,
-					url: response.url,
-					description: 'Bunny CDN purge response',
-				});
-				Logger.error(
-					{
-						status: response.status,
-						responseBodyBytes: errorBody.byteLength,
-						url: formatUrlForDiagnostics(url),
-						label,
-					},
-					'Failed to purge URL via Bunny CDN',
-				);
-				await queue.addUrls([url]);
-				stats.requeued++;
-				continue;
-			}
-			FetchUtils.discardResponseBody(response.body, response.status);
-			stats.purged++;
-		} catch (error) {
-			stats.failures++;
-			Logger.error({error, url: formatUrlForDiagnostics(url), label}, 'Error purging URL via Bunny CDN');
-			await queue.addUrls([url]);
-			stats.requeued++;
-		}
-	}
-	return stats;
-}
-
-const processBunnyPurgeQueue: WorkerTaskHandler = async (_payload, _helpers) => {
-	if (!Config.bunny.purgeEnabled) {
-		Logger.debug('Bunny CDN cache purge is disabled, skipping queue processing');
-		return;
-	}
-	if (!Config.bunny.apiKey || !Config.bunny.pullZoneId) {
-		Logger.error('Bunny CDN cache purge is enabled but credentials are missing');
-		return;
-	}
-	const deps = getWorkerDependencies();
-	const queue = deps.purgeQueue as BunnyPurgeQueue;
-	const apiKey = Config.bunny.apiKey;
-	const totalStats = createEmptyPurgeStats();
-	try {
-		const queueSizeBefore = await queue.getQueueSize();
-		if (queueSizeBefore === 0) {
-			Logger.debug('CDN purge queue is empty');
-			return;
-		}
-		Logger.debug({queueSize: queueSizeBefore}, 'Processing CDN purge queue');
-		const exactBatch = await queue.dequeueExactBatch(EXACT_BATCH_SIZE);
-		if (exactBatch.urls.length > 0) {
-			const exactStats = await purgeUrls(exactBatch.urls, apiKey, queue, 'exact');
-			mergePurgeStats(totalStats, exactStats);
-		}
-		const prefixBatch = await queue.dequeuePrefixBatch(PREFIX_BATCH_SIZE);
-		if (prefixBatch.urls.length > 0) {
-			const prefixStats = await purgeUrls(prefixBatch.urls, apiKey, queue, 'prefix');
-			mergePurgeStats(totalStats, prefixStats);
-		}
-		const remainingQueueSize = await queue.getQueueSize();
-		Logger.debug(
-			{
-				totalAttempted: totalStats.attempted,
-				totalPurged: totalStats.purged,
-				totalRequeued: totalStats.requeued,
-				totalFailures: totalStats.failures,
-				totalRateLimited: totalStats.rateLimited,
-				remainingQueueSize,
-			},
-			'Finished processing CDN purge queue',
-		);
-	} catch (error) {
-		Logger.error({error}, 'Error processing CDN purge queue');
-		throw error;
-	}
-};
-
-export default processBunnyPurgeQueue;

fluxer_api/src/api/worker/tasks/ProcessCachePurgeQueue.ts

@@ -0,0 +1,104 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+import {Config} from '@app/api/Config';
+import {
+	type CachePurgeBatch,
+	type CachePurgeOutcome,
+	createCachePurgeAdapter,
+} from '@app/api/infrastructure/CachePurgeAdapter';
+import {CachePurgeQueue} from '@app/api/infrastructure/CachePurgeQueue';
+import {Logger} from '@app/api/Logger';
+import {getWorkerDependencies} from '@app/api/worker/WorkerContext';
+import type {CachePurgeAdapterName} from '@fluxer/config/src/MasterConfig';
+import type {WorkerTaskHandler} from '@pkgs/worker/src/contracts/WorkerTask';
+
+const RUN_DEADLINE_MS = 10_000;
+
+type CachePurgeFailure = Extract<CachePurgeOutcome, {kind: 'failed'}>;
+
+function describeBatch(adapter: CachePurgeAdapterName, batch: CachePurgeBatch) {
+	return {adapter, exact: batch.exact.length, prefix: batch.prefix.length};
+}
+
+function logFailure(adapter: CachePurgeAdapterName, requeued: CachePurgeBatch, failure: CachePurgeFailure): void {
+	const {status, error} = failure;
+	const context = {...describeBatch(adapter, requeued), status, error};
+	if (status === null || status === 408 || status === 429 || status >= 500) {
+		Logger.warn(context, 'Cache purge request failed, requeued its entries');
+		return;
+	}
+	Logger.error(context, 'Cache purge endpoint refused the request, requeued its entries');
+}
+
+function untriedFrom(singles: ReadonlyArray<CachePurgeBatch>, index: number): CachePurgeBatch {
+	const untried = singles.slice(index);
+	return {exact: untried.flatMap((single) => single.exact), prefix: untried.flatMap((single) => single.prefix)};
+}
+
+const processCachePurgeQueue: WorkerTaskHandler = async (_payload, _helpers) => {
+	const adapterName = Config.cachePurge.adapter;
+	if (adapterName === 'none') {
+		Logger.warn('Skipped a cache purge run because the adapter is none');
+		return;
+	}
+	const adapter = createCachePurgeAdapter(Config.cachePurge);
+	const queue = new CachePurgeQueue(getWorkerDependencies().kvClient);
+	const startedAt = Date.now();
+	const batch = await queue.dequeueBatch();
+	if (batch.exact.length === 0 && batch.prefix.length === 0) {
+		return;
+	}
+	let outcome: CachePurgeOutcome;
+	try {
+		outcome = await adapter.purge(batch);
+	} catch (error) {
+		await queue.requeue(batch);
+		throw error;
+	}
+	if (outcome.kind === 'purged') {
+		Logger.debug(describeBatch(adapterName, batch), 'Purged a batch of cache entries');
+		return;
+	}
+	if (outcome.kind === 'failed') {
+		await queue.requeue(batch);
+		logFailure(adapterName, batch, outcome);
+		return;
+	}
+	Logger.warn(
+		{...describeBatch(adapterName, batch), status: outcome.status},
+		'Cache purge endpoint rejected the batch, retrying each entry on its own',
+	);
+	const singles: Array<CachePurgeBatch> = [
+		...batch.exact.map((entry) => ({exact: [entry], prefix: []})),
+		...batch.prefix.map((entry) => ({exact: [], prefix: [entry]})),
+	];
+	for (const [index, single] of singles.entries()) {
+		if (Date.now() - startedAt >= RUN_DEADLINE_MS) {
+			const untried = untriedFrom(singles, index);
+			await queue.requeue(untried);
+			Logger.warn(
+				describeBatch(adapterName, untried),
+				'Cache purge run reached its deadline, requeued untried entries',
+			);
+			return;
+		}
+		let singleOutcome: CachePurgeOutcome;
+		try {
+			singleOutcome = await adapter.purge(single);
+			if (singleOutcome.kind === 'invalid_entries') {
+				await queue.reject(single);
+			}
+		} catch (error) {
+			await queue.requeue(untriedFrom(singles, index));
+			throw error;
+		}
+		if (singleOutcome.kind === 'failed') {
+			const untried = untriedFrom(singles, index);
+			await queue.requeue(untried);
+			logFailure(adapterName, untried, singleOutcome);
+			return;
+		}
+	}
+};
+
+export default processCachePurgeQueue;

fluxer_api/src/api/worker/tests/ProcessAssetDeletionQueue.test.ts

@@ -1,7 +1,7 @@
 // SPDX-License-Identifier: AGPL-3.0-or-later
 
 import {AssetDeletionQueue} from '@app/api/infrastructure/AssetDeletionQueue';
-import {NoopPurgeQueue} from '@app/api/infrastructure/BunnyPurgeQueue';
+import {NoopPurgeQueue} from '@app/api/infrastructure/CachePurgeQueue';
 import type {IStorageService} from '@app/api/infrastructure/IStorageService';
 import {MockKVProvider} from '@app/api/test/mocks/MockKVProvider';
 import {NoopLogger} from '@app/api/test/mocks/NoopLogger';

fluxer_api/src/api/worker/tests/ProcessCachePurgeQueue.test.ts

@@ -0,0 +1,315 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+import {Config} from '@app/api/Config';
+import type {APICachePurgeConfig} from '@app/api/config/APIConfig';
+import {CachePurgeQueue} from '@app/api/infrastructure/CachePurgeQueue';
+import {MockKVProvider} from '@app/api/test/mocks/MockKVProvider';
+import {NoopLogger} from '@app/api/test/mocks/NoopLogger';
+import {server} from '@app/api/test/msw/server';
+import processCachePurgeQueue from '@app/api/worker/tasks/ProcessCachePurgeQueue';
+import {clearWorkerDependencies, setWorkerDependenciesForTest} from '@app/api/worker/WorkerContext';
+import type {WorkerTaskHelpers} from '@pkgs/worker/src/contracts/WorkerTask';
+import {delay, HttpResponse, http} from 'msw';
+import {afterEach, beforeEach, describe, expect, it, vi} from 'vitest';
+
+const HELPERS = {logger: new NoopLogger()} as unknown as WorkerTaskHelpers;
+const ENDPOINT = 'https://cache-purge.test/purge';
+const MEDIA = 'https://media.test';
+const EXACT_KEY = 'cache_purge:exact';
+const PREFIX_KEY = 'cache_purge:prefix';
+const REJECTED_KEY = 'cache_purge:rejected';
+
+interface PurgeBody {
+	exact: Array<string>;
+	prefix: Array<string>;
+}
+
+interface RecordedRequest {
+	authorization: string | null;
+	contentType: string | null;
+	body: PurgeBody;
+}
+
+function createHarness() {
+	const kvClient = new MockKVProvider();
+	setWorkerDependenciesForTest({kvClient});
+	return {kvClient, queue: new CachePurgeQueue(kvClient)};
+}
+
+function recordPurges(respond: (body: PurgeBody) => Response | Promise<Response>): Array<RecordedRequest> {
+	const requests: Array<RecordedRequest> = [];
+	server.use(
+		http.post(ENDPOINT, async ({request}) => {
+			const body = (await request.json()) as PurgeBody;
+			requests.push({
+				authorization: request.headers.get('authorization'),
+				contentType: request.headers.get('content-type'),
+				body,
+			});
+			return respond(body);
+		}),
+	);
+	return requests;
+}
+
+function entryCount(body: PurgeBody): number {
+	return body.exact.length + body.prefix.length;
+}
+
+async function members(kvClient: MockKVProvider, key: string): Promise<Array<string>> {
+	return (await kvClient.smembers(key)).sort();
+}
+
+function sorted(values: Array<string>): Array<string> {
+	return [...values].sort();
+}
+
+describe('processCachePurgeQueue', () => {
+	let previousCachePurge: APICachePurgeConfig;
+
+	beforeEach(() => {
+		previousCachePurge = {adapter: Config.cachePurge.adapter, http: Config.cachePurge.http};
+		Config.cachePurge.adapter = 'http';
+		Config.cachePurge.http = {endpoint: ENDPOINT, token: 'test-token', timeoutMs: 50};
+	});
+
+	afterEach(() => {
+		Config.cachePurge.adapter = previousCachePurge.adapter;
+		Config.cachePurge.http = previousCachePurge.http;
+		clearWorkerDependencies();
+		vi.useRealTimers();
+	});
+
+	it('sends queued exact and prefix entries in one request and drains the queue', async () => {
+		const {kvClient, queue} = createHarness();
+		const exact = [`${MEDIA}/avatars/1/a_abc`, `${MEDIA}/emojis/9.webp`, `${MEDIA}/attachments/1/2/ação.png`];
+		const prefix = [`${MEDIA}/some/dir/`];
+		await queue.addUrls([...exact, ...prefix]);
+		const requests = recordPurges(() => new HttpResponse(null, {status: 204}));
+
+		await processCachePurgeQueue({}, HELPERS);
+
+		expect(requests).toHaveLength(1);
+		expect(requests[0]!.authorization).toBe('Bearer test-token');
+		expect(requests[0]!.contentType).toBe('application/json');
+		expect(sorted(requests[0]!.body.exact)).toEqual(sorted(exact));
+		expect(requests[0]!.body.prefix).toEqual(prefix);
+		expect(await members(kvClient, EXACT_KEY)).toEqual([]);
+		expect(await members(kvClient, PREFIX_KEY)).toEqual([]);
+	});
+
+	it('strips a trailing asterisk from a prefix entry and keeps a trailing slash', async () => {
+		const {kvClient, queue} = createHarness();
+		await queue.addUrls([`${MEDIA}/stickers/1**`, `${MEDIA}/emojis/`]);
+		const requests = recordPurges(() => new HttpResponse(null, {status: 204}));
+
+		await processCachePurgeQueue({}, HELPERS);
+
+		expect(requests).toHaveLength(1);
+		expect(requests[0]!.body.exact).toEqual([]);
+		expect(sorted(requests[0]!.body.prefix)).toEqual([`${MEDIA}/emojis/`, `${MEDIA}/stickers/1*`]);
+		expect(await members(kvClient, PREFIX_KEY)).toEqual([]);
+	});
+
+	it('sends no authorization header when no token is configured', async () => {
+		Config.cachePurge.http.token = '';
+		const {queue} = createHarness();
+		await queue.addUrls([`${MEDIA}/avatars/1/abc`]);
+		const requests = recordPurges(() => new HttpResponse(null, {status: 204}));
+
+		await processCachePurgeQueue({}, HELPERS);
+
+		expect(requests).toHaveLength(1);
+		expect(requests[0]!.authorization).toBeNull();
+	});
+
+	it('requeues the whole batch after a server error', async () => {
+		const {kvClient, queue} = createHarness();
+		const exact = [`${MEDIA}/avatars/1/abc`, `${MEDIA}/banners/1/def`];
+		const prefix = [`${MEDIA}/some/dir/`];
+		await queue.addUrls([...exact, ...prefix]);
+		const requests = recordPurges(() => new HttpResponse(null, {status: 503}));
+
+		await processCachePurgeQueue({}, HELPERS);
+
+		expect(requests).toHaveLength(1);
+		expect(await members(kvClient, EXACT_KEY)).toEqual(sorted(exact));
+		expect(await members(kvClient, PREFIX_KEY)).toEqual(prefix);
+	});
+
+	it('requeues the batch when the endpoint does not answer in time', async () => {
+		const {kvClient, queue} = createHarness();
+		const exact = [`${MEDIA}/avatars/1/abc`, `${MEDIA}/banners/1/def`];
+		await queue.addUrls(exact);
+		const requests = recordPurges(async () => {
+			await delay('infinite');
+			return new HttpResponse(null, {status: 204});
+		});
+
+		await processCachePurgeQueue({}, HELPERS);
+
+		expect(requests).toHaveLength(1);
+		expect(await members(kvClient, EXACT_KEY)).toEqual(sorted(exact));
+	});
+
+	it('requeues the batch when the endpoint is unreachable', async () => {
+		const {kvClient, queue} = createHarness();
+		const exact = [`${MEDIA}/avatars/1/abc`, `${MEDIA}/banners/1/def`];
+		await queue.addUrls(exact);
+		const requests = recordPurges(() => HttpResponse.error());
+
+		await processCachePurgeQueue({}, HELPERS);
+
+		expect(requests).toHaveLength(1);
+		expect(await members(kvClient, EXACT_KEY)).toEqual(sorted(exact));
+	});
+
+	it('requeues the batch on a redirect without following it', async () => {
+		const {kvClient, queue} = createHarness();
+		const exact = [`${MEDIA}/avatars/1/abc`, `${MEDIA}/banners/1/def`];
+		await queue.addUrls(exact);
+		const redirectTarget = 'https://cache-purge.test/elsewhere';
+		const redirectedRequests: Array<string> = [];
+		server.use(
+			http.all(redirectTarget, ({request}) => {
+				redirectedRequests.push(request.method);
+				return new HttpResponse(null, {status: 204});
+			}),
+		);
+		const requests = recordPurges(() => new HttpResponse(null, {status: 302, headers: {Location: redirectTarget}}));
+
+		await processCachePurgeQueue({}, HELPERS);
+
+		expect(requests).toHaveLength(1);
+		expect(redirectedRequests).toEqual([]);
+		expect(await members(kvClient, EXACT_KEY)).toEqual(sorted(exact));
+	});
+
+	it('requeues the batch on an authorisation failure without splitting it', async () => {
+		const {kvClient, queue} = createHarness();
+		const exact = [`${MEDIA}/avatars/1/abc`, `${MEDIA}/banners/1/def`, `${MEDIA}/emojis/9.webp`];
+		await queue.addUrls(exact);
+		const requests = recordPurges(() => new HttpResponse(null, {status: 401}));
+
+		await processCachePurgeQueue({}, HELPERS);
+
+		expect(requests).toHaveLength(1);
+		expect(await members(kvClient, EXACT_KEY)).toEqual(sorted(exact));
+		expect(await members(kvClient, REJECTED_KEY)).toEqual([]);
+	});
+
+	it('sets aside only the entry the endpoint rejects on its own', async () => {
+		const {kvClient, queue} = createHarness();
+		const poison = `${MEDIA}/attachments/1/2/poison.png`;
+		await queue.addUrls([`${MEDIA}/avatars/1/abc`, poison, `${MEDIA}/emojis/9.webp`]);
+		const requests = recordPurges((body) =>
+			body.exact.includes(poison) ? new HttpResponse(null, {status: 422}) : new HttpResponse(null, {status: 204}),
+		);
+
+		await processCachePurgeQueue({}, HELPERS);
+
+		expect(requests.map((request) => entryCount(request.body))).toEqual([3, 1, 1, 1]);
+		expect(await members(kvClient, REJECTED_KEY)).toEqual([poison]);
+		expect(await members(kvClient, EXACT_KEY)).toEqual([]);
+	});
+
+	it('splits the batch when the endpoint answers 400', async () => {
+		const {kvClient, queue} = createHarness();
+		const poison = `${MEDIA}/attachments/1/2/poison.png`;
+		await queue.addUrls([`${MEDIA}/avatars/1/abc`, poison, `${MEDIA}/emojis/9.webp`]);
+		const requests = recordPurges((body) =>
+			body.exact.includes(poison) ? new HttpResponse(null, {status: 400}) : new HttpResponse(null, {status: 204}),
+		);
+
+		await processCachePurgeQueue({}, HELPERS);
+
+		expect(requests.map((request) => entryCount(request.body))).toEqual([3, 1, 1, 1]);
+		expect(await members(kvClient, REJECTED_KEY)).toEqual([poison]);
+		expect(await members(kvClient, EXACT_KEY)).toEqual([]);
+	});
+
+	it('keeps untried exact and prefix entries queued when a single-entry retry fails', async () => {
+		const {kvClient, queue} = createHarness();
+		const first = `${MEDIA}/avatars/1/abc`;
+		const second = `${MEDIA}/banners/1/def`;
+		const third = `${MEDIA}/emojis/9.webp`;
+		const directory = `${MEDIA}/some/dir/`;
+		await queue.addUrls([first, second, third, directory]);
+		let singleRequests = 0;
+		const requests = recordPurges((body) => {
+			if (entryCount(body) > 1) {
+				return new HttpResponse(null, {status: 422});
+			}
+			singleRequests++;
+			return new HttpResponse(null, {status: singleRequests === 1 ? 204 : 503});
+		});
+
+		await processCachePurgeQueue({}, HELPERS);
+
+		expect(requests.map((request) => request.body.exact)).toEqual([[first, second, third], [first], [second]]);
+		expect(await members(kvClient, EXACT_KEY)).toEqual(sorted([second, third]));
+		expect(await members(kvClient, PREFIX_KEY)).toEqual([directory]);
+		expect(await members(kvClient, REJECTED_KEY)).toEqual([]);
+	});
+
+	it('requeues untried entries when the fallback runs out of time', async () => {
+		vi.useFakeTimers({toFake: ['Date']});
+		vi.setSystemTime(new Date('2026-09-13T00:00:00.000Z'));
+		const {kvClient, queue} = createHarness();
+		const entries = [1, 2, 3, 4, 5].map((index) => `${MEDIA}/avatars/${index}/abc`);
+		await queue.addUrls(entries);
+		const requests = recordPurges((body) => {
+			if (entryCount(body) > 1) {
+				return new HttpResponse(null, {status: 422});
+			}
+			vi.setSystemTime(Date.now() + 4_000);
+			return new HttpResponse(null, {status: 204});
+		});
+
+		await processCachePurgeQueue({}, HELPERS);
+
+		expect(requests.map((request) => request.body.exact)).toEqual([entries, [entries[0]], [entries[1]], [entries[2]]]);
+		expect(await members(kvClient, EXACT_KEY)).toEqual(sorted([entries[3]!, entries[4]!]));
+		expect(await members(kvClient, REJECTED_KEY)).toEqual([]);
+	});
+
+	it('leaves the queue untouched when the adapter is none', async () => {
+		Config.cachePurge.adapter = 'none';
+		const {kvClient, queue} = createHarness();
+		const exact = [`${MEDIA}/avatars/1/abc`];
+		const prefix = [`${MEDIA}/some/dir/`];
+		await queue.addUrls([...exact, ...prefix]);
+		const requests = recordPurges(() => new HttpResponse(null, {status: 204}));
+
+		await processCachePurgeQueue({}, HELPERS);
+
+		expect(requests).toEqual([]);
+		expect(await members(kvClient, EXACT_KEY)).toEqual(exact);
+		expect(await members(kvClient, PREFIX_KEY)).toEqual(prefix);
+		expect(await kvClient.get('cache_purge:budget:exact')).toBeNull();
+		expect(await kvClient.get('cache_purge:budget:prefix')).toBeNull();
+	});
+
+	it('sends no more than the token bucket allows and refills at the configured rate', async () => {
+		vi.useFakeTimers({toFake: ['Date']});
+		vi.setSystemTime(new Date('2026-09-13T00:00:00.000Z'));
+		const {kvClient, queue} = createHarness();
+		await queue.addUrls([
+			...Array.from({length: 200}, (_, index) => `${MEDIA}/avatars/${index}/abc`),
+			...Array.from({length: 30}, (_, index) => `${MEDIA}/dirs/${index}/`),
+		]);
+		const requests = recordPurges(() => new HttpResponse(null, {status: 204}));
+
+		await processCachePurgeQueue({}, HELPERS);
+		await processCachePurgeQueue({}, HELPERS);
+		vi.setSystemTime(Date.now() + 10_000);
+		await processCachePurgeQueue({}, HELPERS);
+
+		expect(requests.map((request) => [request.body.exact.length, request.body.prefix.length])).toEqual([
+			[120, 20],
+			[50, 5],
+		]);
+		expect(await kvClient.scard(EXACT_KEY)).toBe(30);
+		expect(await kvClient.scard(PREFIX_KEY)).toBe(5);
+	});
+});