From 3dc344be65cbba2d44b6db9af7167dbff41d6704 Mon Sep 17 00:00:00 2001 From: Hampus Date: Mon, 31 Aug 2026 22:26:21 +0200 Subject: [PATCH] fix(worker): keep attachment decay state on a stale expiry row (#2285) --- .../attachment/AttachmentDecayRepository.ts | 22 ++++- .../src/api/worker/tasks/ExpireAttachments.ts | 4 +- .../worker/tests/ExpireAttachments.test.ts | 91 +++++++++++++++++++ 3 files changed, 114 insertions(+), 3 deletions(-) create mode 100644 fluxer_api/src/api/worker/tests/ExpireAttachments.test.ts diff --git a/fluxer_api/src/api/attachment/AttachmentDecayRepository.ts b/fluxer_api/src/api/attachment/AttachmentDecayRepository.ts index eba08c58e..f5a929b46 100644 --- a/fluxer_api/src/api/attachment/AttachmentDecayRepository.ts +++ b/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 { + 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 { const batch = new BatchBuilder(); batch.addPrepared( diff --git a/fluxer_api/src/api/worker/tasks/ExpireAttachments.ts b/fluxer_api/src/api/worker/tasks/ExpireAttachments.ts index 392279627..afa63aa1f 100644 --- a/fluxer_api/src/api/worker/tasks/ExpireAttachments.ts +++ b/fluxer_api/src/api/worker/tasks/ExpireAttachments.ts @@ -29,7 +29,7 @@ export async function processExpiredAttachments(now = new Date()): Promise 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 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, diff --git a/fluxer_api/src/api/worker/tests/ExpireAttachments.test.ts b/fluxer_api/src/api/worker/tests/ExpireAttachments.test.ts new file mode 100644 index 000000000..eb04ca8c7 --- /dev/null +++ b/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} { + const queued: Array = []; + const queue = { + async queueDeletion(item: Omit) { + 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 { + 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(); + }); +});