fix: run handleDelete finds and prepares inside the writer lock

app/containers/MessageErrorActions.test.tsx

@@ -0,0 +1,117 @@
+import { createRef } from 'react';
+import { act, render } from '@testing-library/react-native';
+
+import MessageErrorActions, { type IMessageErrorActions } from './MessageErrorActions';
+import database from '../lib/database';
+import log from '../lib/methods/helpers/log';
+import { type TMessageModel } from '../definitions';
+import { FakeDatabase, type FakeModel } from '../lib/database/__tests__/mockedWatermelonDB';
+
+jest.mock('../lib/database', () => ({
+	__esModule: true,
+	default: { active: null }
+}));
+
+jest.mock('../lib/methods/helpers/log', () => ({
+	__esModule: true,
+	default: jest.fn()
+}));
+
+jest.mock('../lib/methods/sendMessage', () => ({
+	resendMessage: jest.fn(() => Promise.resolve())
+}));
+
+const mockShowActionSheet = jest.fn();
+jest.mock('./ActionSheet', () => ({
+	useActionSheet: () => ({ showActionSheet: mockShowActionSheet })
+}));
+
+const renderComponent = (tmid?: string) => {
+	const ref = createRef<IMessageErrorActions>();
+	render(<MessageErrorActions ref={ref} tmid={tmid} />);
+	return ref;
+};
+
+const pressDelete = (ref: React.RefObject<IMessageErrorActions | null>, message: FakeModel) => {
+	ref.current?.showMessageErrorActions(message as unknown as TMessageModel);
+	const { options } = mockShowActionSheet.mock.calls[mockShowActionSheet.mock.calls.length - 1][0];
+	const deleteOption = options.find((o: { icon: string }) => o.icon === 'delete');
+	return deleteOption.onPress() as Promise<void>;
+};
+
+let db: FakeDatabase;
+
+beforeEach(() => {
+	jest.clearAllMocks();
+	db = new FakeDatabase();
+	(database as unknown as { active: FakeDatabase }).active = db;
+});
+
+describe('MessageErrorActions handleDelete', () => {
+	it('deletes a failed thread message while another writer touches the same record', async () => {
+		const threadMessage = db.add('thread_messages', 'msg-1');
+		const message = db.add('messages', 'msg-1');
+		db.add('messages', 'tmid-1', { tcount: 1, tlm: new Date() });
+		db.add('threads', 'tmid-1');
+
+		const ref = renderComponent('tmid-1');
+
+		let concurrentWriter: Promise<unknown> = Promise.resolve();
+		await act(async () => {
+			const deleting = pressDelete(ref, threadMessage);
+			// A saga writing to the very same record while the delete is in flight
+			concurrentWriter = db.write(async () => {
+				await db.batch([threadMessage.prepareUpdate(m => (m.tcount = 0))]);
+			});
+			await Promise.all([deleting, concurrentWriter]);
+		});
+
+		await expect(concurrentWriter).resolves.not.toThrow();
+		expect(log).not.toHaveBeenCalled();
+
+		// no prepare and no find escaped the writer lock
+		expect(db.prepareLog.every(entry => entry.insideWriter)).toBe(true);
+		expect(db.findLog.every(entry => entry.insideWriter)).toBe(true);
+
+		// the whole tree was committed in a single batch
+		expect(db.batches[0]).toEqual([
+			'thread_messages#msg-1:destroyPermanently',
+			'messages#msg-1:destroyPermanently',
+			'messages#tmid-1:update',
+			'threads#tmid-1:destroyPermanently'
+		]);
+		// the thread header lost its thread count and the thread record is gone
+		expect(db.collections.messages.get('tmid-1')?.tcount).toBeNull();
+		expect(db.collections.messages.get('tmid-1')?.tlm).toBeNull();
+		expect(message._preparedState).toBeNull();
+	});
+
+	it('decrements the thread count when other messages remain', async () => {
+		const threadMessage = db.add('thread_messages', 'msg-1');
+		db.add('messages', 'tmid-1', { tcount: 3 });
+		db.add('threads', 'tmid-1');
+
+		const ref = renderComponent('tmid-1');
+		await act(async () => {
+			await pressDelete(ref, threadMessage);
+		});
+
+		expect(log).not.toHaveBeenCalled();
+		expect(db.collections.messages.get('tmid-1')?.tcount).toBe(2);
+		expect(db.batches[0]).toEqual(['thread_messages#msg-1:destroyPermanently', 'messages#tmid-1:update']);
+	});
+
+	it('destroys only the message on the non-thread branch', async () => {
+		const message = db.add('messages', 'msg-1');
+
+		const ref = renderComponent();
+		await act(async () => {
+			await pressDelete(ref, message);
+		});
+
+		expect(log).not.toHaveBeenCalled();
+		expect(db.findLog).toEqual([]);
+		expect(db.batches).toEqual([['messages#msg-1:destroyPermanently']]);
+		expect(db.prepareLog).toEqual([{ record: 'messages#msg-1', op: 'destroyPermanently', insideWriter: true }]);
+	});
+});

app/containers/MessageErrorActions.tsx

@@ -23,54 +23,55 @@ const MessageErrorActions = forwardRef<IMessageErrorActions, { tmid?: string }>(
 	const handleDelete = async (message: TMessageModel) => {
 		try {
 			const db = database.active;
-			const deleteBatch: Model[] = [];
 			const msgCollection = db.get('messages');
 			const threadCollection = db.get('threads');
 
-			// Delete the object (it can be Message or ThreadMessage instance)
-			deleteBatch.push(message.prepareDestroyPermanently());
+			await db.write(async () => {
+				const deleteBatch: Model[] = [];
 
-			// If it's a thread, we find and delete the whole tree, if necessary
-			if (tmid) {
-				try {
-					const msg = await msgCollection.find(message.id);
-					deleteBatch.push(msg.prepareDestroyPermanently());
-				} catch {
-					// Do nothing: message not found
-				}
+				// Delete the object (it can be Message or ThreadMessage instance)
+				deleteBatch.push(message.prepareDestroyPermanently());
 
-				try {
-					// Find the thread header and update it
-					const msg = await msgCollection.find(tmid);
-					if (msg?.tcount && msg.tcount <= 1) {
-						deleteBatch.push(
-							msg.prepareUpdate(m => {
-								m.tcount = null;
-								m.tlm = null;
-							})
-						);
+				// If it's a thread, we find and delete the whole tree, if necessary
+				if (tmid) {
+					try {
+						const msg = await msgCollection.find(message.id);
+						deleteBatch.push(msg.prepareDestroyPermanently());
+					} catch {
+						// Do nothing: message not found
+					}
 
-						try {
-							// If the whole thread was removed, delete the thread
-							const thread = await threadCollection.find(tmid);
-							deleteBatch.push(thread.prepareDestroyPermanently());
-						} catch {
-							// Do nothing: thread not found
+					try {
+						// Find the thread header and update it
+						const msg = await msgCollection.find(tmid);
+						if (msg?.tcount && msg.tcount <= 1) {
+							deleteBatch.push(
+								msg.prepareUpdate(m => {
+									m.tcount = null;
+									m.tlm = null;
+								})
+							);
+
+							try {
+								// If the whole thread was removed, delete the thread
+								const thread = await threadCollection.find(tmid);
+								deleteBatch.push(thread.prepareDestroyPermanently());
+							} catch {
+								// Do nothing: thread not found
+							}
+						} else {
+							deleteBatch.push(
+								msg.prepareUpdate(m => {
+									if (m.tcount) {
+										m.tcount -= 1;
+									}
+								})
+							);
 						}
-					} else {
-						deleteBatch.push(
-							msg.prepareUpdate(m => {
-								if (m.tcount) {
-									m.tcount -= 1;
-								}
-							})
-						);
+					} catch {
+						// Do nothing: message not found
 					}
-				} catch {
-					// Do nothing: message not found
 				}
-			}
-			await db.write(async () => {
 				await db.batch(deleteBatch);
 			});
 		} catch (e) {

app/lib/database/__tests__/mockedWatermelonDB.tsx

@@ -0,0 +1,168 @@
+// WatermelonDB stand-ins: pending changes, batch inside a writer, and the serialized writer lock.
+
+export const tick = () => Promise.resolve();
+
+// Let every already-queued microtask/promise chain settle.
+export const flush = () => new Promise(resolve => setImmediate(resolve));
+
+export const deferred = () => {
+	let resolve: () => void = () => undefined;
+	const promise = new Promise<void>(r => {
+		resolve = r;
+	});
+	return { promise, resolve };
+};
+
+export interface IFakeRecord {
+	_preparedState: string | null;
+	prepareUpdate: (updater?: (record: any) => void) => any;
+	prepareDestroyPermanently: () => any;
+	[key: string]: any;
+}
+
+export interface ILogEntry {
+	record: string;
+	op: string;
+	insideWriter: boolean;
+}
+
+interface IFakeModelHost {
+	insideWriter: boolean;
+	prepareLog: ILogEntry[];
+}
+
+export class FakeModel {
+	id: string;
+	table: string;
+	_preparedState: string | null = null;
+	_db?: IFakeModelHost;
+	[key: string]: any;
+
+	constructor(table: string, id: string, props: Record<string, any> = {}, db?: IFakeModelHost) {
+		this._db = db;
+		this.table = table;
+		this.id = id;
+		Object.assign(this, props);
+	}
+
+	get _debugName() {
+		return `${this.table}#${this.id}`;
+	}
+
+	_log(op: string) {
+		this._db?.prepareLog.push({ record: this._debugName, op, insideWriter: this._db.insideWriter });
+	}
+
+	prepareDestroyPermanently() {
+		this._log('destroyPermanently');
+		if (this._preparedState) {
+			throw new Error(`Cannot destroy permanently record with pending changes (${this._debugName})`);
+		}
+		this._preparedState = 'destroyPermanently';
+		return this;
+	}
+
+	prepareUpdate(updater: (m: any) => void = () => {}) {
+		this._log('update');
+		if (this._preparedState) {
+			throw new Error(`Cannot update a record with pending changes (${this._debugName})`);
+		}
+		updater(this);
+		this._preparedState = 'update';
+		return this;
+	}
+}
+
+// A record identified by a `table#id` debug name, for tests that don't need a whole FakeDatabase.
+export const makeFakeRecord = (debugName: string, fields: Record<string, any> = {}): IFakeRecord => {
+	const [table, id] = debugName.split('#');
+	return new FakeModel(table, id, fields) as unknown as IFakeRecord;
+};
+
+// Serialized writer lock, like WatermelonDB's.
+export const createWriterLock = () => {
+	let queue: Promise<unknown> = Promise.resolve();
+	return <T,>(work: () => Promise<T>): Promise<T> => {
+		const run = queue.then(() => work());
+		queue = run.catch(() => undefined);
+		return run;
+	};
+};
+
+// db.batch commits prepared records, clearing their pending state (like the real writer).
+export const commitPreparedRecords = (...args: any[]) => {
+	args.flat().forEach((item: any) => {
+		if (item && typeof item === 'object' && '_preparedState' in item) {
+			item._preparedState = null;
+		}
+	});
+	return Promise.resolve(undefined);
+};
+
+export const createBatchMock = () => jest.fn(commitPreparedRecords);
+
+// Whether the mocked `log` received a WatermelonDB "pending changes" error.
+export const loggedPendingChanges = (log: unknown) =>
+	(log as jest.Mock).mock.calls.some(([error]) => /pending changes/.test(error?.message ?? ''));
+
+// An instrumented `database.active` that records what each writer prepared, found and batched.
+export class FakeDatabase {
+	insideWriter = false;
+	prepareLog: ILogEntry[] = [];
+	findLog: { record: string; insideWriter: boolean }[] = [];
+	batches: string[][] = [];
+	collections: Record<string, Map<string, FakeModel>> = {
+		messages: new Map(),
+		threads: new Map(),
+		thread_messages: new Map()
+	};
+	_queue = createWriterLock();
+
+	add(table: string, id: string, props: Record<string, any> = {}) {
+		const record = new FakeModel(table, id, props, this);
+		if (!this.collections[table]) {
+			this.collections[table] = new Map();
+		}
+		this.collections[table].set(id, record);
+		return record;
+	}
+
+	get(table: string) {
+		return {
+			find: async (id: string) => {
+				this.findLog.push({ record: `${table}#${id}`, insideWriter: this.insideWriter });
+				await tick();
+				const record = this.collections[table]?.get(id);
+				if (!record) {
+					throw new Error(`Record ${table}#${id} not found`);
+				}
+				return record;
+			}
+		};
+	}
+
+	write<T>(work: () => Promise<T>): Promise<T> {
+		return this._queue(async () => {
+			this.insideWriter = true;
+			try {
+				return await work();
+			} finally {
+				this.insideWriter = false;
+			}
+		});
+	}
+
+	async batch(records: FakeModel[]) {
+		if (!this.insideWriter) {
+			throw new Error('Database.batch() can only be called from inside of a Writer');
+		}
+		await tick();
+		this.batches.push(records.map(r => `${r._debugName}:${r._preparedState}`));
+		records.forEach(record => {
+			if (!record._preparedState) {
+				throw new Error("Cannot batch a record that doesn't have a prepared create/update/delete");
+			}
+			record._preparedState = null;
+		});
+	}
+}