From cdf54836822c75bbe5e42c37210e69b48341721c Mon Sep 17 00:00:00 2001 From: OtavioStasiak Date: Tue, 4 Aug 2026 17:00:37 -0300 Subject: [PATCH 1/4] fix(db): move decryptPendingMessages prepares inside the writer lock --- app/lib/encryption/encryption.test.ts | 119 ++++++++++++++++++++++++-- app/lib/encryption/encryption.ts | 32 ++++--- 2 files changed, 131 insertions(+), 20 deletions(-) diff --git a/app/lib/encryption/encryption.test.ts b/app/lib/encryption/encryption.test.ts index 18a9d9f0a01..f8762fc0115 100644 --- a/app/lib/encryption/encryption.test.ts +++ b/app/lib/encryption/encryption.test.ts @@ -1,6 +1,7 @@ // Bypass the global mock of `app/lib/encryption` declared in jest.setup.js by // importing directly from the file. We exercise the real `encryptMessage` here. import encryption from './encryption'; +import database from '../database'; jest.unmock('./encryption'); @@ -48,14 +49,38 @@ jest.mock('../store/auxStore', () => ({ })); const mockSubFind = jest.fn(); -jest.mock('../database', () => ({ - __esModule: true, - default: { - active: { - get: () => ({ find: (rid: string) => mockSubFind(rid) }) +// Rows returned by `collection.query(...).fetch()`, keyed by collection name. +const mockQueryRows: Record = {}; +const mockDbBatch = jest.fn((...args: any[]) => { + // db.batch commits prepared records, clearing their pending state (like the real writer). + args.flat().forEach((item: any) => { + if (item && typeof item === 'object' && '_preparedState' in item) { + item._preparedState = null; } - } -})); + }); + return Promise.resolve(undefined); +}); +jest.mock('../database', () => { + let writerQueue: Promise = Promise.resolve(); + return { + __esModule: true, + default: { + active: { + get: (name: string) => ({ + find: (rid: string) => mockSubFind(rid), + query: () => ({ fetch: () => Promise.resolve(mockQueryRows[name] ?? []) }) + }), + // Serialized writer lock, like WatermelonDB's. + write: (callback: () => Promise) => { + const run = writerQueue.then(() => callback()); + writerQueue = run.catch(() => undefined); + return run; + }, + batch: (...args: unknown[]) => mockDbBatch(...args) + } + } + }; +}); const mockRoomEncrypt = jest.fn(); const mockHasSessionKey = jest.fn(); @@ -141,3 +166,83 @@ describe('Encryption.encryptMessage', () => { expect(mockRoomEncrypt).not.toHaveBeenCalled(); }); }); + +describe('Encryption.decryptPendingMessages', () => { + const rid = 'r1'; + + // Mimics a WatermelonDB Model: prepareUpdate throws while a previous prepared + // update has not been committed yet. + const makeMessageRecord = (id: string) => { + const record: any = { + id, + t: 'e2e', + msg: 'cipher', + subscription: { id: rid }, + _preparedState: null as string | null, + prepareUpdate(recordUpdater: (m: any) => void) { + if (record._preparedState) { + throw new Error(`Cannot update a record with pending changes (messages#${id})`); + } + recordUpdater(record); + record._preparedState = 'update'; + return record; + } + }; + return record; + }; + + const deferred = () => { + let resolve: () => void = () => undefined; + const promise = new Promise(r => { + resolve = r; + }); + return { promise, resolve }; + }; + + beforeEach(() => { + jest.clearAllMocks(); + mockQueryRows.messages = []; + mockQueryRows.threads = []; + mockQueryRows.thread_messages = []; + }); + + afterEach(() => { + jest.restoreAllMocks(); + }); + + it('does not throw "pending changes" when a concurrent writer touches the same message mid-decrypt', async () => { + const record = makeMessageRecord('m1'); + mockQueryRows.messages = [record]; + jest.spyOn(encryption, 'decryptMessage').mockResolvedValue({ msg: 'plain', e2e: 'done' } as any); + + const db = (database as any).active; + + // Hold the writer lock — as a new incoming message being saved would — and then update + // the very record decryptPendingMessages is about to prepare. + const concurrentGate = deferred(); + const concurrentWrite = db.write(async () => { + await concurrentGate.promise; + await db.batch([ + record.prepareUpdate((m: any) => { + m.msg = 'written by another writer'; + }) + ]); + }); + + const decrypting = encryption.decryptPendingMessages(rid); + + // Give an unlocked implementation the chance to prepare now — before the concurrent + // writer runs — and hold the record pending until its own batch. + await new Promise(resolve => setImmediate(resolve)); + concurrentGate.resolve(); + + await expect(Promise.all([concurrentWrite, decrypting])).resolves.toBeDefined(); + + // The decrypted update reached db.batch and nothing was left prepared-but-uncommitted. + const committed = mockDbBatch.mock.calls.map(call => call.flat()).some(items => items.includes(record)); + expect(committed).toBe(true); + expect(record.msg).toBe('plain'); + expect(record.e2e).toBe('done'); + expect(record._preparedState).toBeNull(); + }); +}); diff --git a/app/lib/encryption/encryption.ts b/app/lib/encryption/encryption.ts index ffe13c69655..00ba90c1225 100644 --- a/app/lib/encryption/encryption.ts +++ b/app/lib/encryption/encryption.ts @@ -335,12 +335,13 @@ class Encryption { const threadMessagesToDecrypt = await threadMessagesCollection.query(...whereClause).fetch(); // Concat messages/threads/threadMessages - let toDecrypt: (TThreadModel | TThreadMessageModel | TMessageModel)[] = [ + const toDecrypt: (TThreadModel | TThreadMessageModel | TMessageModel)[] = [ ...messagesToDecrypt, ...threadsToDecrypt, ...threadMessagesToDecrypt ]; - toDecrypt = (await Promise.all( + + const decrypted = await Promise.all( toDecrypt.map(async message => { const { t, msg, tmsg, attachments, content } = message; let newMessage: Partial = {}; @@ -357,20 +358,25 @@ class Encryption { } as IMessage); } - try { - return message.prepareUpdate( - protectedFunction((m: TMessageModel) => { - Object.assign(m, newMessage); - }) - ); - } catch { - return null; - } + return { message, newMessage }; }) - )) as (TThreadModel | TThreadMessageModel)[]; + ); await db.write(async () => { - await db.batch(toDecrypt); + const prepared = decrypted + .map(({ message, newMessage }) => { + try { + return message.prepareUpdate( + protectedFunction((m: TMessageModel) => { + Object.assign(m, newMessage); + }) + ); + } catch { + return null; + } + }) + .filter((record): record is TThreadModel | TThreadMessageModel => record !== null); + await db.batch(prepared); }); } catch (e) { log(e); From 00120dad0afcbb455ef3f43ed39203badae8f050 Mon Sep 17 00:00:00 2001 From: OtavioStasiak Date: Fri, 14 Aug 2026 13:48:20 -0300 Subject: [PATCH 2/4] code improvements --- app/lib/encryption/encryption.ts | 16 ++++++++++------ 1 file changed, 10 insertions(+), 6 deletions(-) diff --git a/app/lib/encryption/encryption.ts b/app/lib/encryption/encryption.ts index 00ba90c1225..1dcd3d56ae2 100644 --- a/app/lib/encryption/encryption.ts +++ b/app/lib/encryption/encryption.ts @@ -345,8 +345,8 @@ class Encryption { toDecrypt.map(async message => { const { t, msg, tmsg, attachments, content } = message; let newMessage: Partial = {}; - if (message.subscription) { - const { id: rid } = message.subscription; + const rid = message.subscription?.id; + if (rid) { // WM Object -> Plain Object newMessage = await this.decryptMessage({ t, @@ -362,6 +362,10 @@ class Encryption { }) ); + if (!decrypted.length) { + return; + } + await db.write(async () => { const prepared = decrypted .map(({ message, newMessage }) => { @@ -371,12 +375,12 @@ class Encryption { Object.assign(m, newMessage); }) ); - } catch { + } catch (e) { + log(e); return null; } - }) - .filter((record): record is TThreadModel | TThreadMessageModel => record !== null); - await db.batch(prepared); + }); + await db.batch(...prepared); }); } catch (e) { log(e); From c711f0357260c7f32db4ce00c9fb22700de5b008 Mon Sep 17 00:00:00 2001 From: OtavioStasiak Date: Fri, 14 Aug 2026 13:48:41 -0300 Subject: [PATCH 3/4] chore: new test case encryption --- app/lib/encryption/encryption.test.ts | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) diff --git a/app/lib/encryption/encryption.test.ts b/app/lib/encryption/encryption.test.ts index f8762fc0115..bfc2c312c71 100644 --- a/app/lib/encryption/encryption.test.ts +++ b/app/lib/encryption/encryption.test.ts @@ -245,4 +245,21 @@ describe('Encryption.decryptPendingMessages', () => { expect(record.e2e).toBe('done'); expect(record._preparedState).toBeNull(); }); + + it('skips a record whose prepareUpdate throws and still commits the others', async () => { + const failing = makeMessageRecord('m1'); + // Already prepared by someone else, so prepareUpdate throws for this one. + failing._preparedState = 'update'; + const healthy = makeMessageRecord('m2'); + mockQueryRows.messages = [failing, healthy]; + jest.spyOn(encryption, 'decryptMessage').mockResolvedValue({ msg: 'plain', e2e: 'done' } as any); + + await encryption.decryptPendingMessages(rid); + + const batched = mockDbBatch.mock.calls.flatMap(call => call.flat()); + expect(batched).toContain(healthy); + expect(batched).not.toContain(failing); + expect(healthy.msg).toBe('plain'); + expect(failing.msg).toBe('cipher'); + }); }); From ef2149313804dc964425f1582bf8175366390b99 Mon Sep 17 00:00:00 2001 From: OtavioStasiak Date: Fri, 14 Aug 2026 16:49:37 +0000 Subject: [PATCH 4/4] chore: format code and fix lint issues --- app/lib/encryption/encryption.ts | 25 ++++++++++++------------- 1 file changed, 12 insertions(+), 13 deletions(-) diff --git a/app/lib/encryption/encryption.ts b/app/lib/encryption/encryption.ts index 1dcd3d56ae2..d87708656db 100644 --- a/app/lib/encryption/encryption.ts +++ b/app/lib/encryption/encryption.ts @@ -367,19 +367,18 @@ class Encryption { } await db.write(async () => { - const prepared = decrypted - .map(({ message, newMessage }) => { - try { - return message.prepareUpdate( - protectedFunction((m: TMessageModel) => { - Object.assign(m, newMessage); - }) - ); - } catch (e) { - log(e); - return null; - } - }); + const prepared = decrypted.map(({ message, newMessage }) => { + try { + return message.prepareUpdate( + protectedFunction((m: TMessageModel) => { + Object.assign(m, newMessage); + }) + ); + } catch (e) { + log(e); + return null; + } + }); await db.batch(...prepared); }); } catch (e) {