From e704115b568b5f58b8440059593f4b07202ccc0e Mon Sep 17 00:00:00 2001 From: OtavioStasiak Date: Tue, 4 Aug 2026 20:12:17 -0300 Subject: [PATCH] fix(db): move deleteMessage finds and prepares inside the writer lock --- app/lib/methods/subscriptions/room.test.ts | 110 +++++++++++++++++++++ app/lib/methods/subscriptions/room.ts | 54 +++++----- 2 files changed, 137 insertions(+), 27 deletions(-) diff --git a/app/lib/methods/subscriptions/room.test.ts b/app/lib/methods/subscriptions/room.test.ts index bf0b0e9576d..005a31242c4 100644 --- a/app/lib/methods/subscriptions/room.test.ts +++ b/app/lib/methods/subscriptions/room.test.ts @@ -1,6 +1,9 @@ +import { InteractionManager } from 'react-native'; + import RoomSubscription from './room'; import { getMessageById } from '../../database/services/Message'; import { getThreadById } from '../../database/services/Thread'; +import database from '../../database'; import log from '../helpers/log'; const mockSubscribeRoom = jest.fn, [string]>(() => Promise.resolve([])); @@ -174,4 +177,111 @@ describe('RoomSubscription', () => { expect(loggedPendingChanges).toBe(false); }); }); + + describe('deleteMessage concurrency', () => { + // Mimics a WatermelonDB Model: both prepare* calls throw while a previous + // prepared change has not been committed yet. + const makeDeletableRecord = (debugName: string) => { + const record: any = { + _preparedState: null as string | null, + prepareUpdate(recordUpdater: (m: any) => void) { + if (record._preparedState) { + throw new Error(`Cannot update a record with pending changes (${debugName})`); + } + recordUpdater(record); + record._preparedState = 'update'; + return record; + }, + prepareDestroyPermanently() { + if (record._preparedState) { + throw new Error(`Cannot destroy permanently record with pending changes (${debugName})`); + } + record._preparedState = 'destroyPermanently'; + return record; + } + }; + return record; + }; + + const deferred = () => { + let resolve: () => void = () => undefined; + const promise = new Promise(r => { + resolve = r; + }); + return { promise, resolve }; + }; + + let interactionTask: Promise | null = null; + + beforeEach(() => { + interactionTask = null; + // Run the deferred work inline so the test can await it. + jest.spyOn(InteractionManager, 'runAfterInteractions').mockImplementation((task: any) => { + interactionTask = task(); + return { then: () => undefined, done: () => undefined, cancel: () => undefined } as any; + }); + mockDbBatch.mockImplementation((...items: any[]) => { + items.flat().forEach(item => { + if (item && typeof item === 'object' && '_preparedState' in item) { + item._preparedState = null; + } + }); + return Promise.resolve(undefined); + }); + }); + + afterEach(() => { + jest.restoreAllMocks(); + }); + + it('does not throw "pending changes" when a concurrent writer touches a message being deleted', async () => { + const _id = 'KXse45i7gGYE8j4Xb'; + const messageRecord = makeDeletableRecord(`messages#${_id}`); + const threadRecord = makeDeletableRecord(`threads#${_id}`); + const threadMessageRecord = makeDeletableRecord(`thread_messages#${_id}`); + const collections: Record = { + messages: { find: () => Promise.resolve(messageRecord) }, + threads: { find: () => Promise.resolve(threadRecord) }, + thread_messages: { find: () => Promise.resolve(threadMessageRecord) } + }; + mockDbGet.mockImplementation((name: string) => collections[name]); + + const db = (database as any).active; + + // Hold the writer lock — as an incoming message update would — and then touch the very + // record the delete branch is about to prepare for destruction. + const concurrentGate = deferred(); + const concurrentWrite = db.write(async () => { + await concurrentGate.promise; + await db.batch([ + messageRecord.prepareUpdate((m: any) => { + m.msg = 'written by another writer'; + }) + ]); + }); + + await sub.handleNotifyRoomReceived({ + fields: { eventName: `${rid}/deleteMessage`, args: [{ _id }] } + } as any); + + // Give an unlocked implementation the chance to prepare now — before the concurrent + // writer runs — and hold the records pending until its own batch. + await new Promise(resolve => setImmediate(resolve)); + concurrentGate.resolve(); + + await expect(Promise.all([concurrentWrite, interactionTask])).resolves.toBeDefined(); + + const loggedPendingChanges = (log as jest.Mock).mock.calls.some(([err]) => /pending changes/.test(err?.message)); + expect(loggedPendingChanges).toBe(false); + + // The whole delete batch committed together and nothing was left prepared. + const deleteBatch = mockDbBatch.mock.calls + .map(call => call.flat()) + .find(items => items.includes(threadRecord) && items.includes(threadMessageRecord)); + expect(deleteBatch).toContain(messageRecord); + expect(messageRecord._preparedState).toBeNull(); + expect(threadRecord._preparedState).toBeNull(); + expect(threadMessageRecord._preparedState).toBeNull(); + }); + }); }); diff --git a/app/lib/methods/subscriptions/room.ts b/app/lib/methods/subscriptions/room.ts index acfe1c74029..33dd6b46047 100644 --- a/app/lib/methods/subscriptions/room.ts +++ b/app/lib/methods/subscriptions/room.ts @@ -148,34 +148,34 @@ export default class RoomSubscription { const msgCollection = db.get('messages'); const threadsCollection = db.get('threads'); const threadMessagesCollection = db.get('thread_messages'); - let deleteMessage: TMessageModel; - let deleteThread: TThreadModel; - let deleteThreadMessage: TThreadMessageModel; - - // Delete message - try { - const m = await msgCollection.find(_id); - deleteMessage = m.prepareDestroyPermanently(); - } catch (e) { - // Do nothing - } - - // Delete thread - try { - const m = await threadsCollection.find(_id); - deleteThread = m.prepareDestroyPermanently(); - } catch (e) { - // Do nothing - } - - // Delete thread message - try { - const m = await threadMessagesCollection.find(_id); - deleteThreadMessage = m.prepareDestroyPermanently(); - } catch (e) { - // Do nothing - } await db.write(async () => { + let deleteMessage: TMessageModel | undefined; + let deleteThread: TThreadModel | undefined; + let deleteThreadMessage: TThreadMessageModel | undefined; + + // Delete message + try { + const m = await msgCollection.find(_id); + deleteMessage = m.prepareDestroyPermanently(); + } catch (e) { + // Do nothing + } + + // Delete thread + try { + const m = await threadsCollection.find(_id); + deleteThread = m.prepareDestroyPermanently(); + } catch (e) { + // Do nothing + } + + // Delete thread message + try { + const m = await threadMessagesCollection.find(_id); + deleteThreadMessage = m.prepareDestroyPermanently(); + } catch (e) { + // Do nothing + } await db.batch(deleteMessage, deleteThread, deleteThreadMessage); }); } catch (e) {