fix(worker): keep attachment decay state on a stale expiry row (#2285)

fluxer_api/src/api/attachment/AttachmentDecayRepository.ts

@@ -1,7 +1,13 @@
 // SPDX-License-Identifier: AGPL-3.0-or-later
 
 import type {AttachmentID, ChannelID, MessageID} from '../BrandedTypes';
-import {BatchBuilder, fetchMany, fetchManyInChunks, fetchOne} from '../database/CassandraQueryExecution';
+import {
+	BatchBuilder,
+	deleteOneOrMany,
+	fetchMany,
+	fetchManyInChunks,
+	fetchOne,
+} from '../database/CassandraQueryExecution';
 import {AttachmentDecayByExpiry, AttachmentDecayById} from '../Tables';
 import type {AttachmentDecayRow} from '../types/AttachmentDecayTypes';
 
@@ -73,6 +79,20 @@ export class AttachmentDecayRepository {
 		return fetchMany(query.bind({expiry_bucket: bucket, current_time: currentTime}));
 	}
 
+	async deleteExpiryRecord(params: {
+		expiry_bucket: number;
+		expires_at: Date;
+		attachment_id: AttachmentID;
+	}): Promise<void> {
+		await deleteOneOrMany(
+			AttachmentDecayByExpiry.deleteByPk({
+				expiry_bucket: params.expiry_bucket,
+				expires_at: params.expires_at,
+				attachment_id: params.attachment_id,
+			}),
+		);
+	}
+
 	async deleteRecords(params: {expiry_bucket: number; expires_at: Date; attachment_id: AttachmentID}): Promise<void> {
 		const batch = new BatchBuilder();
 		batch.addPrepared(

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

@@ -29,7 +29,7 @@ export async function processExpiredAttachments(now = new Date()): Promise<void>
 			for (const row of expired) {
 				const metadata = await repo.fetchById(row.attachment_id);
 				if (!metadata) {
-					await repo.deleteRecords({
+					await repo.deleteExpiryRecord({
 						expiry_bucket: row.expiry_bucket,
 						expires_at: row.expires_at,
 						attachment_id: row.attachment_id,
@@ -38,7 +38,7 @@ export async function processExpiredAttachments(now = new Date()): Promise<void>
 					continue;
 				}
 				if (metadata.expires_at > row.expires_at) {
-					await repo.deleteRecords({
+					await repo.deleteExpiryRecord({
 						expiry_bucket: row.expiry_bucket,
 						expires_at: row.expires_at,
 						attachment_id: row.attachment_id,

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

@@ -0,0 +1,91 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+import {ms} from 'itty-time';
+import {afterEach, describe, expect, it} from 'vitest';
+import {AttachmentDecayRepository} from '../../attachment/AttachmentDecayRepository';
+import {createAttachmentID, createChannelID, createMessageID} from '../../BrandedTypes';
+import type {IAssetDeletionQueue, QueuedAssetDeletion} from '../../infrastructure/IAssetDeletionQueue';
+import type {InstanceConfigRepository} from '../../instance/InstanceConfigRepository';
+import {getExpiryBucket} from '../../utils/AttachmentDecay';
+import {processExpiredAttachments} from '../tasks/ExpireAttachments';
+import {clearWorkerDependencies, setWorkerDependenciesForTest} from '../WorkerContext';
+
+const ATTACHMENT_ID = createAttachmentID(9001n);
+const CHANNEL_ID = createChannelID(9002n);
+const MESSAGE_ID = createMessageID(9003n);
+const FILENAME = 'decaying.png';
+const UPLOADED_AT = new Date('2026-01-01T00:00:00.000Z');
+const FIRST_EXPIRY = new Date('2026-01-31T00:00:00.000Z');
+const EXTENDED_EXPIRY = new Date('2026-02-25T00:00:00.000Z');
+
+function createQueue(): {queue: IAssetDeletionQueue; queued: Array<QueuedAssetDeletion>} {
+	const queued: Array<QueuedAssetDeletion> = [];
+	const queue = {
+		async queueDeletion(item: Omit<QueuedAssetDeletion, 'queuedAt' | 'retryCount'>) {
+			queued.push({...item});
+		},
+	} as unknown as IAssetDeletionQueue;
+	return {queue, queued};
+}
+
+function installDependencies(queue: IAssetDeletionQueue): void {
+	setWorkerDependenciesForTest({
+		assetDeletionQueue: queue,
+		instanceConfigRepository: {
+			async getEffectiveAttachmentDecayConfig() {
+				return {enabled: true};
+			},
+		} as unknown as InstanceConfigRepository,
+	});
+}
+
+async function writeDecayRecord(repository: AttachmentDecayRepository, expiresAt: Date): Promise<void> {
+	await repository.upsert({
+		attachment_id: ATTACHMENT_ID,
+		channel_id: CHANNEL_ID,
+		message_id: MESSAGE_ID,
+		filename: FILENAME,
+		size_bytes: 1024n,
+		uploaded_at: UPLOADED_AT,
+		expires_at: expiresAt,
+		last_accessed_at: UPLOADED_AT,
+		cost: 1,
+		lifetime_days: 30,
+		status: null,
+		expiry_bucket: getExpiryBucket(expiresAt),
+	});
+}
+
+describe('processExpiredAttachments', () => {
+	afterEach(() => {
+		clearWorkerDependencies();
+	});
+
+	it('keeps the decay record when it clears a superseded expiry row', async () => {
+		const repository = new AttachmentDecayRepository();
+		const {queue} = createQueue();
+		installDependencies(queue);
+		await writeDecayRecord(repository, FIRST_EXPIRY);
+		await writeDecayRecord(repository, EXTENDED_EXPIRY);
+
+		await processExpiredAttachments(new Date(FIRST_EXPIRY.getTime() + ms('1 day')));
+
+		const record = await repository.fetchById(ATTACHMENT_ID);
+		expect(record?.expires_at.toISOString()).toBe(EXTENDED_EXPIRY.toISOString());
+		expect(await repository.fetchExpiredByBucket(getExpiryBucket(FIRST_EXPIRY), EXTENDED_EXPIRY, 10)).toHaveLength(0);
+	});
+
+	it('still queues the asset once the extended expiry passes', async () => {
+		const repository = new AttachmentDecayRepository();
+		const {queue, queued} = createQueue();
+		installDependencies(queue);
+		await writeDecayRecord(repository, FIRST_EXPIRY);
+		await writeDecayRecord(repository, EXTENDED_EXPIRY);
+
+		await processExpiredAttachments(new Date(FIRST_EXPIRY.getTime() + ms('1 day')));
+		await processExpiredAttachments(new Date(EXTENDED_EXPIRY.getTime() + ms('1 day')));
+
+		expect(queued.map((item) => item.s3Key)).toEqual([`attachments/${CHANNEL_ID}/${ATTACHMENT_ID}/${FILENAME}`]);
+		expect(await repository.fetchById(ATTACHMENT_ID)).toBeNull();
+	});
+});