From a05dbdec2736d3227d118c0ba41355ae633da7e2 Mon Sep 17 00:00:00 2001 From: Diego Mello Date: Mon, 3 Aug 2026 18:11:58 -0300 Subject: [PATCH] fix: prepare subscription updates inside the writer lock createOrUpdateSubscription and decryptPendingSubscriptions called prepareUpdate outside db.write and committed the batch later. A concurrent updateLastOpen took the writer lock in that gap, called update() on the same cached subscription record, and threw "Cannot update a record with pending changes". Both paths now prepare and batch inside one db.write, as room.ts already does. --- app/lib/encryption/encryption.ts | 41 +++-- app/lib/methods/subscriptions/rooms.test.ts | 174 ++++++++++++++++++++ app/lib/methods/subscriptions/rooms.ts | 123 +++++++------- 3 files changed, 261 insertions(+), 77 deletions(-) create mode 100644 app/lib/methods/subscriptions/rooms.test.ts diff --git a/app/lib/encryption/encryption.ts b/app/lib/encryption/encryption.ts index ffe13c69655..186ccd7d0ff 100644 --- a/app/lib/encryption/encryption.ts +++ b/app/lib/encryption/encryption.ts @@ -1,4 +1,4 @@ -import { type Model, Q } from '@nozbe/watermelondb'; +import { Q } from '@nozbe/watermelondb'; import EJSON from 'ejson'; import { deleteAsync } from 'expo-file-system/legacy'; import { @@ -395,25 +395,32 @@ class Encryption { sub => sub.lastMessage?.t === E2E_MESSAGE_TYPE && sub.lastMessage?.e2e !== E2E_STATUS.DONE ); - const preparedSubscriptions: (Model | null)[] = await Promise.all( - subsEncryptedToDecrypt.map(async (sub: TSubscriptionModel) => { - const newSub = await this.decryptSubscription(sub); - try { - return sub.prepareUpdate( - protectedFunction((m: TSubscriptionModel) => { - if (newSub?.lastMessage) { - m.lastMessage = newSub.lastMessage; - } - }) - ); - } catch { - return null; - } - }) + const decryptedSubscriptions = await Promise.all( + subsEncryptedToDecrypt.map(async (sub: TSubscriptionModel) => ({ + sub, + newSub: await this.decryptSubscription(sub) + })) ); + // Prepare and batch under the writer lock so a concurrent writer can't + // call prepareUpdate on a record with pending changes. await db.write(async () => { - await db.batch(preparedSubscriptions.filter((record): record is Model => record !== null)); + const preparedSubscriptions = decryptedSubscriptions + .map(({ sub, newSub }) => { + try { + return sub.prepareUpdate( + protectedFunction((m: TSubscriptionModel) => { + if (newSub?.lastMessage) { + m.lastMessage = newSub.lastMessage; + } + }) + ); + } catch { + return null; + } + }) + .filter((record): record is TSubscriptionModel => record !== null); + await db.batch(preparedSubscriptions); }); } catch (e) { log(e); diff --git a/app/lib/methods/subscriptions/rooms.test.ts b/app/lib/methods/subscriptions/rooms.test.ts new file mode 100644 index 00000000000..13d5274e413 --- /dev/null +++ b/app/lib/methods/subscriptions/rooms.test.ts @@ -0,0 +1,174 @@ +import { createOrUpdateSubscription } from './rooms'; +import { updateLastOpen } from '../updateLastOpen'; +import { getSubscriptionByRoomId } from '../../database/services/Subscription'; +import { getMessageById } from '../../database/services/Message'; +import log from '../helpers/log'; + +jest.mock('../../services/sdk', () => ({ + __esModule: true, + default: {} +})); + +jest.mock('../../store/auxStore', () => ({ + store: { + getState: jest.fn(() => ({ room: { subscribedRoom: null } })), + dispatch: jest.fn() + } +})); + +jest.mock('../helpers/log', () => ({ + __esModule: true, + default: jest.fn() +})); + +jest.mock('../helpers', () => ({ + getRoomAvatar: jest.fn(), + getRoomTitle: jest.fn(), + getSenderName: jest.fn(), + random: jest.fn() +})); + +jest.mock('../helpers/protectedFunction', () => ({ + __esModule: true, + default: (fn: (...args: unknown[]) => unknown) => fn +})); + +jest.mock('../helpers/buildMessage', () => ({ + __esModule: true, + default: (msg: unknown) => msg +})); + +jest.mock('../helpers/mergeSubscriptionsRooms', () => ({ + merge: (subscription: unknown) => subscription +})); + +jest.mock('../../encryption', () => ({ + Encryption: { + decryptPendingSubscriptions: jest.fn(), + decryptPendingMessages: jest.fn(), + getRoomInstance: jest.fn() + } +})); + +jest.mock('../updateMessages', () => ({ + __esModule: true, + default: jest.fn() +})); + +jest.mock('../getRoom', () => ({ + getRoom: jest.fn() +})); + +jest.mock('../actions', () => ({ + handlePayloadUserInteraction: jest.fn() +})); + +jest.mock('../../../actions/room', () => ({ + removedRoom: jest.fn() +})); + +jest.mock('../../../actions/login', () => ({ + setUser: jest.fn() +})); + +jest.mock('../../../actions/videoConf', () => ({ + handleVideoConfIncomingWebsocketMessages: jest.fn() +})); + +jest.mock('../../../containers/InAppNotification', () => ({ + INAPP_NOTIFICATION_EMITTER: 'NotificationInApp' +})); + +const mockDbBatch = jest.fn(); +jest.mock('../../database', () => { + let writerQueue: Promise = Promise.resolve(); + const mockCollection = { + find: jest.fn(() => Promise.reject(new Error('not found'))), + prepareCreate: jest.fn(() => ({})), + schema: {} + }; + return { + __esModule: true, + default: { + active: { + get: () => mockCollection, + write: jest.fn((callback: () => Promise) => { + const run = writerQueue.then(() => callback()); + writerQueue = run.catch(() => undefined); + return run; + }), + batch: (...args: unknown[]) => mockDbBatch(...args) + } + } + }; +}); + +jest.mock('../../database/services/Subscription', () => ({ + getSubscriptionByRoomId: jest.fn() +})); + +jest.mock('../../database/services/Message', () => ({ + getMessageById: jest.fn() +})); + +const rid = 'GENERAL'; + +// Mimics a WatermelonDB Model: one cached instance per record, and +// prepareUpdate throws while a previous prepared update is not committed. +const makeSubscriptionRecord = () => { + const record: any = { + rid, + lastOpen: null, + _preparedState: null as string | null, + prepareUpdate(recordUpdater: (s: any) => void) { + if (record._preparedState) { + throw new Error(`Cannot update a record with pending changes (subscriptions#${rid})`); + } + recordUpdater(record); + record._preparedState = 'update'; + return record; + }, + update(recordUpdater: (s: any) => void) { + record.prepareUpdate(recordUpdater); + record._preparedState = null; + return Promise.resolve(record); + } + }; + return record; +}; + +describe('createOrUpdateSubscription concurrency', () => { + beforeEach(() => { + jest.clearAllMocks(); + mockDbBatch.mockImplementation((batch: any[]) => { + (Array.isArray(batch) ? batch : [batch]).forEach(item => { + if (item && typeof item === 'object' && '_preparedState' in item) { + item._preparedState = null; + } + }); + return Promise.resolve(undefined); + }); + }); + + it('does not leave a prepared subscription visible to a concurrent updateLastOpen', async () => { + const record = makeSubscriptionRecord(); + (getSubscriptionByRoomId as jest.Mock).mockResolvedValue(record); + // Slow message lookup keeps createOrUpdateSubscription busy after it fetched the subscription. + (getMessageById as jest.Mock).mockImplementation(() => new Promise(resolve => setTimeout(() => resolve(null), 10))); + + const subscription = { + rid, + _id: rid, + lastMessage: { _id: 'msg-id', rid, msg: 'hi' } + } as any; + + await Promise.all([ + createOrUpdateSubscription(subscription, undefined as any), + updateLastOpen(rid, [{ _updatedAt: '2026-01-01T12:00:00.000Z' }]) + ]); + + const loggedPendingChanges = (log as jest.Mock).mock.calls.some(([error]) => /pending changes/.test(error?.message)); + expect(loggedPendingChanges).toBe(false); + expect(record.lastOpen).toEqual(new Date('2026-01-01T12:00:00.000Z')); + }); +}); diff --git a/app/lib/methods/subscriptions/rooms.ts b/app/lib/methods/subscriptions/rooms.ts index 78aeb9ca674..0501a2b3552 100644 --- a/app/lib/methods/subscriptions/rooms.ts +++ b/app/lib/methods/subscriptions/rooms.ts @@ -46,7 +46,7 @@ const WINDOW_TIME = 500; export let roomsSubscription: { stop: () => void } | null = null; -const createOrUpdateSubscription = async (subscription: ISubscription, room: IServerRoom | IRoom) => { +export const createOrUpdateSubscription = async (subscription: ISubscription, room: IServerRoom | IRoom) => { try { const db = database.active; const subCollection = db.get('subscriptions'); @@ -150,74 +150,77 @@ const createOrUpdateSubscription = async (subscription: ISubscription, room: ISe } const tmp = merge(subscription, room); - const sub = await getSubscriptionByRoomId(tmp.rid); - const batch: Model[] = []; - if (sub) { - try { - const update = sub.prepareUpdate(s => { - Object.assign(s, tmp); - if (subscription.announcement) { - if (subscription.announcement !== sub.announcement) { - s.bannerClosed = false; + // Serialize the fetch, prepares and the batch under the writer lock so a concurrent + // writer can't call prepareUpdate on a record with pending changes. + await db.write(async () => { + const sub = await getSubscriptionByRoomId(tmp.rid); + + const batch: Model[] = []; + if (sub) { + try { + const update = sub.prepareUpdate(s => { + Object.assign(s, tmp); + if (subscription.announcement) { + if (subscription.announcement !== sub.announcement) { + s.bannerClosed = false; + } } - } - if (sub.hideUnreadStatus && subscription.hasOwnProperty('hideUnreadStatus')) { - if (sub.hideUnreadStatus !== subscription.hideUnreadStatus) { - s.hideUnreadStatus = !!subscription.hideUnreadStatus; + if (sub.hideUnreadStatus && subscription.hasOwnProperty('hideUnreadStatus')) { + if (sub.hideUnreadStatus !== subscription.hideUnreadStatus) { + s.hideUnreadStatus = !!subscription.hideUnreadStatus; + } } - } - }); - batch.push(update); - } catch (e) { - console.log(e); - } - } else { - try { - const create = subCollection.prepareCreate(s => { - s._raw = sanitizedRaw({ id: tmp.rid }, subCollection.schema); - Object.assign(s, tmp); - if (s.roomUpdatedAt) { - s.roomUpdatedAt = new Date(); - } - }); - batch.push(create); - } catch (e) { - console.log(e); + }); + batch.push(update); + } catch (e) { + console.log(e); + } + } else { + try { + const create = subCollection.prepareCreate(s => { + s._raw = sanitizedRaw({ id: tmp.rid }, subCollection.schema); + Object.assign(s, tmp); + if (s.roomUpdatedAt) { + s.roomUpdatedAt = new Date(); + } + }); + batch.push(create); + } catch (e) { + console.log(e); + } } - } - const { subscribedRoom } = store.getState().room; - if (tmp.lastMessage && subscribedRoom !== tmp.rid) { - const lastMessage = buildMessage(tmp.lastMessage); - const messagesCollection = db.get('messages'); - let messageRecord = {} as TMessageModel | null; - if (lastMessage) { - messageRecord = await getMessageById(lastMessage._id); - } + const { subscribedRoom } = store.getState().room; + if (tmp.lastMessage && subscribedRoom !== tmp.rid) { + const lastMessage = buildMessage(tmp.lastMessage); + const messagesCollection = db.get('messages'); + let messageRecord = {} as TMessageModel | null; + if (lastMessage) { + messageRecord = await getMessageById(lastMessage._id); + } - if (messageRecord) { - batch.push( - messageRecord.prepareUpdate(() => { - Object.assign(messageRecord, lastMessage); - }) - ); - } else { - batch.push( - messagesCollection.prepareCreate(m => { - if (lastMessage) { - m._raw = sanitizedRaw({ id: lastMessage._id }, messagesCollection.schema); - if (m.subscription) { - m.subscription.id = lastMessage.rid; + if (messageRecord) { + batch.push( + messageRecord.prepareUpdate(() => { + Object.assign(messageRecord, lastMessage); + }) + ); + } else { + batch.push( + messagesCollection.prepareCreate(m => { + if (lastMessage) { + m._raw = sanitizedRaw({ id: lastMessage._id }, messagesCollection.schema); + if (m.subscription) { + m.subscription.id = lastMessage.rid; + } } - } - return Object.assign(m, lastMessage); - }) - ); + return Object.assign(m, lastMessage); + }) + ); + } } - } - await db.write(async () => { await db.batch(batch); });