fix: prepare subscription updates inside the writer lock #7544
+261
−77
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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<unknown> = 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<void>) => { | ||
| 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')); | ||
| }); | ||
|
Comment on lines
+153
to
+173
Contributor
There was a problem hiding this comment. 📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win 🧩 Analysis chain🏁 Script executed: #!/bin/bash
# Description: Check whether the new test fails on the pre-fix implementation of createOrUpdateSubscription.
set -euo pipefail
fd -t f 'rooms.ts' app/lib/methods/subscriptions --exec sh -c '
echo "=== $1 (current) ==="
rg -n "db.write|prepareUpdate|prepareCreate|db.batch|getSubscriptionByRoomId|getMessageById" "$1"
' _ {}
echo "=== previous revision of rooms.ts ==="
git log --oneline -3 -- app/lib/methods/subscriptions/rooms.ts
git show HEAD~1:app/lib/methods/subscriptions/rooms.ts 2>/dev/null \
| rg -n "db.write|prepareUpdate|prepareCreate|db.batch|getSubscriptionByRoomId|getMessageById" || echo "previous revision not available"Repository: RocketChat/Rocket.Chat.ReactNative Length of output: 2005 🏁 Script executed: #!/bin/bash
set -euo pipefail
echo "=== current implementation ==="
sed -n '150,228p' app/lib/methods/subscriptions/rooms.ts
echo "=== previous implementation ==="
git show HEAD~1:app/lib/methods/subscriptions/rooms.ts | sed -n '150,228p'
echo "=== test ==="
sed -n '130,180p' app/lib/methods/subscriptions/rooms.test.ts
echo "=== mockDbBatch / db.write definitions in test file ==="
rg -n "mockDbBatch|db\.write|createOrUpdateSubscription|updateLastOpen" app/lib/methods/subscriptions/rooms.test.ts
echo "=== imports and setup ==="
sed -n '1,80p' app/lib/methods/subscriptions/rooms.test.tsRepository: RocketChat/Rocket.Chat.ReactNative Length of output: 8595 🏁 Script executed: #!/bin/bash
set -euo pipefail
python3 - <<'PY'
import asyncio
class MockDb:
def __init__(self):
self.lock = None
self.batch_calls = []
async def write(self, fn):
while self.lock is not None:
await asyncio.sleep(0)
self.lock = True
try:
return await fn()
finally:
self.lock = None
def batch(self, batch):
self.batch_calls.append(batch); return asyncio.sleep(0)
class SubscriptionRecord:
def __init__(self):
self._preparedState = 'pending'
self.lastOpen = 'old'
def prepareUpdate(self, update):
if self._preparedState is None:
raise Exception('RecordAlreadyPrepared: pending changes')
update(self)
return self
def update(self, data):
self.lastOpen = data.get('lastOpen', self.lastOpen)
async def update_last_open_old(db, rid, records):
sub = SubscriptionRecord()
await db.write(lambda: (
db.batch(sub.prepareUpdate(lambda s: s.update(records[0]))),
)[0])
return sub.lastOpen
async def create_or_update_old(db, rid, msg_delay=False):
sub = SubscriptionRecord()
batch = []
batch.append(sub.prepareUpdate(lambda s: s.update({'lastMessage': {'_id': 'msg-id'}})))
if msg_delay:
await asyncio.sleep(0.01)
await db.write(lambda: db.batch(batch))
async def main():
result = []
for name, cfunc in [('old', create_or_update_old), ('old', update_last_open_old)]:
pass
for name, cfunc in [('createOrUpdateSubscription_old', create_or_update_old), ('updateLastOpen_old', update_last_open_old)]:
print(name)
async def run():
db = MockDb()
try:
await asyncio.gather(cfunc(db, 'rid'), update_last_open_old(db, 'rid', [{'lastOpen': 'new'}]))
print('ok')
except Exception as e:
print(f'raised: {e}')
asyncio.run(run())
print('non-deterministic interleaving example: updateLastOpen acquires writer before createOrUpdateSubscription prepares')
async def run():
db = MockDb()
sub = SubscriptionRecord()
batch = []
batch.append(sub.prepareUpdate(lambda s: s.update({'lastMessage': {'_id': 'msg-id'}})))
await asyncio.sleep(0)
await db.write(lambda: db.batch(batch)).get() if hasattr((tmp := db.write(lambda: db.batch(batch))).get, '__await__') else None
PYRepository: RocketChat/Rocket.Chat.ReactNative Length of output: 420 🏁 Script executed: #!/bin/bash
set -euo pipefail
echo "=== updateLastOpen implementation / mocks ==="
rg -n "function updateLastOpen|const updateLastOpen|updateLastOpen|db\.write|prepareUpdate|_preparedState" app/lib -g '*.ts' -g '*.tsx'
echo "=== focused updateLastOpen files ==="
fd -t f '^updateLastOpen\.(ts|tsx)$' app/lib
while IFS= read -r f; do
echo "=== $f ==="
sed -n '1,220p' "$f"
done < <(fd -t f '^updateLastOpen\.(ts|tsx)$' app/lib)Repository: RocketChat/Rocket.Chat.ReactNative Length of output: 11238 🏁 Script executed: #!/bin/bash
set -euo pipefail
echo "=== updateLastOpen implementation ==="
sed -n '1,80p' app/lib/methods/updateLastOpen.ts
echo "=== Subscription service implementation ==="
sed -n '1,180p' app/lib/database/services/Subscription.ts | sed -n '1,160p'Repository: RocketChat/Rocket.Chat.ReactNative Length of output: 2515 Make the concurrency ordering explicit.
🤖 Prompt for AI Agents |
||
| }); | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy lift
🧩 Analysis chain
🏁 Script executed:
Repository: RocketChat/Rocket.Chat.ReactNative
Length of output: 224
🏁 Script executed:
Repository: RocketChat/Rocket.Chat.ReactNative
Length of output: 50391
🏁 Script executed:
Repository: RocketChat/Rocket.Chat.ReactNative
Length of output: 18472
🌐 Web query:
WatermelonDB v0.28.1 db.write prepareUpdate concurrent writes lock writer callbacks💡 Result:
In WatermelonDB, database writes are strictly serialized to ensure data integrity [1][2]. The library employs a writer-locking mechanism where only one writer (an action defined via
database.write()or a@writermethod) can execute at any given time [1][2]. Other incoming write requests are queued until the active writer completes [3]. RegardingprepareUpdateand concurrent writes, note the following critical technical constraints: 1. Writer Exclusivity: Because only one writer can be active at once, concurrent write conflicts are prevented at the library level [1][2]. You do not need to implement manual locks for database consistency, as the queueing system manages this automatically [3][2]. 2. Synchronous Batching Requirement: The methodprepareUpdate(along withprepareCreate,prepareMarkAsDeleted, etc.) must be executed and passed todatabase.batch()synchronously within the same writer context [1][4][5]. Performing asynchronous operations (likeawait) between preparing an update and executing it inbatch()can lead to errors such as: "record.prepareUpdate was called on... but wasn't sent to batch synchronously -- this is bad!" [4][6]. 3. Correct Pattern: All logically connected read and write operations should be wrapped together inside a singledatabase.write()block or an@writermethod [1][2]. Any necessary asynchronous data fetching or calculations should be completed before preparing the operations, or the batching must occur within the same asynchronous writer block where the preparations are made [4][5]. If you need to nest writer operations, you must usecallWriterto ensure the nested operation runs within the existing writer's exclusive lock [1][3]. Nesting writers withoutcallWriterwill result in errors or deadlocks because the system expects only one active writer [1].Citations:
record.prepareUpdate was called on ${this.table}#${this.id} but wasn't sent to batch() synchronously -- this is bad!Nozbe/WatermelonDB#1368🌐 Web query:
WatermelonDB v0.28.1 Source Database/index.js writer lock prepareUpdate callback💡 Result:
In WatermelonDB, the prepareUpdate method is used to prepare an update operation on a model record, but it does not execute the update immediately [1]. To ensure consistency and performance, WatermelonDB requires that these prepared operations are executed synchronously within a single database batch transaction [2][1]. If you receive an error stating that record.prepareUpdate was called but was not sent to batch synchronously, it indicates that the prepared update was created inside a writer but not passed to database.batch within the same execution context [2]. Key points regarding the writer lock and prepareUpdate: 1. Writer Locks: WatermelonDB enforces that all database modifications must occur within a Writer [3][4]. Only one Writer can execute at a time to ensure a consistent view of the database [3]. 2. Synchronous Requirement: When you use methods like prepareUpdate or prepareCreate, you are creating a "prepared" record [1]. These must be passed to database.batch or writer.batch synchronously [2][1]. 3. Correct Usage: You should perform your prepare calls inside the same writer block where you call the batch function [2]. Example of correct usage: await database.write(async => { const record = await collection.find('id'); const preparedUpdate = record.prepareUpdate(model => { model.field = 'new value'; }); await database.batch(preparedUpdate); }); If you perform asynchronous operations (e.g., fetching other data) between preparing an update and batching it, you risk breaking the synchronous requirement [2]. All necessary data should be prepared or fetched before the final batch execution [5].
Citations:
🌐 Web query:
@nozbe/watermelondb 0.28.1-0 GitHub source write writer lock pendingQueue changeRecords💡 Result:
The version 0.28.1-0 of
@nozbe/watermelondbwas released on July 24, 2025 [1][2]. This release is primarily a version bump [1] and does not contain specific changes to the underlying architecture of writer locks, the pending queue, or changeRecords handling. In WatermelonDB, the core concurrency model remains unchanged in this version: 1. Writer Locks and PendingQueue: WatermelonDB enforces a strict concurrency model where only one writer can execute at a time [3]. When multiple write operations are initiated, they are placed in a pending queue [4]. If a writer is already active, subsequent writers are queued [4]. A common diagnostic warning occurs when many writers are queued, which can occur if operations are not batched efficiently [4][5]. 2. ChangeRecords: The concept of changeRecords is intrinsic to WatermelonDB's reactivity and synchronization system, where database mutations (create, update, delete) are tracked to propagate changes to observers and the synchronization engine [6]. 3. Development Recommendations: To avoid issues with writer locks and pending queues, the official documentation recommends batching database writes using database.batch rather than executing many individual writers [3][5]. For nested operations, the library requires the use of callWriter or callReader to manage context correctly within existing transactions [3][4]. For detailed implementation logic or migration guides, developers should refer to the standard documentation for the 0.28.x release series, as no breaking structural changes were introduced in 0.28.1-0 [7][8].Citations:
🏁 Script executed:
Repository: RocketChat/Rocket.Chat.ReactNative
Length of output: 16619
🌐 Web query:
"Nozbe/WatermelonDB" "v0.28.1-0" "src/Database/index.js" "writer" "pendingQueue"💡 Result:
In WatermelonDB, including version 0.28.1-0, database operations are managed by a queuing system to ensure data consistency [1][2]. The code mentioned in src/Database/index.js interacts with an ActionQueue (or WorkQueue in later versions) which handles the scheduling of readers and writers [1][3]. The pending queue system functions as follows: 1. Consistency and Locking: WatermelonDB enforces that only one writer can execute at a time [2]. All write operations (create, update, delete) must occur within a writer block [1][2]. When a write operation is initiated, it is added to a pending queue if another action is currently running [1][2]. 2. Handling Queued Actions: If multiple write operations are triggered simultaneously, they are placed in a queue to be executed sequentially [4]. The library provides warnings if a large number of actions accumulate in this queue, as this may indicate that previous actions are stuck or that the application is inefficiently dispatching individual writes instead of using batching [5][4]. 3. Nesting: If you need to perform a write operation from within another writer, you must use the callWriter method [2]. Direct execution of a nested writer will fail because the database lock is already held by the parent writer [2]. 4. Internal Mechanics: Internally, the Database class maintains an instance of the queue (often referenced as _actionQueue or _workQueue in the source) [1][3]. Methods like unsafeResetDatabase specifically trigger an abort of all pending actions to ensure a clean state during reset operations [1]. If you are encountering issues related to a large number of writers in the queue, the recommended practice is to aggregate changes into a single batch operation using db.batch instead of executing multiple individual writes [4][2].
Citations:
Guard against stale decrypted subscription snapshots.
decryptSubscription(lastMessage)runs beforedb.write. If another writer changes the same subscription between decryption andprepareUpdate, this callback can overwrite the newerlastMessagewith the older decrypted value.Capture the source last-message identity or timestamp before decryption. Validate it against the current record inside
prepareUpdate; if it changed, re-read and decrypt the current record or skip and schedule a retry. Add a regression test for a newerlastMessagearriving during delayed subscription decryption.🤖 Prompt for AI Agents