Skip to content

fix: leaked prepared subscriptions break lastOpen writes #7626

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
diegolmello wants to merge 3 commits into
base: develop
Choose a base branch
from
Open

fix: leaked prepared subscriptions break lastOpen writes #7626

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
c067356
fix: leaked prepared subscriptions break lastOpen writes
diegolmello Sep 1, 2026
1afb9c2
fix: guard last-message write on the built message, skip empty subscr…
diegolmello Sep 1, 2026
d66d4f9
fix: unblock sibling subscriptions when one decryption rejects
diegolmello Sep 1, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
76 changes: 56 additions & 20 deletions app/lib/encryption/encryption.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -167,29 +167,29 @@ describe('Encryption.encryptMessage', () => {
});
});

// Mimics a WatermelonDB Model: prepareUpdate throws while a previous prepared
// update has not been committed yet.
const makeRecord = (table: string, id: string, fields: Record<string, any> = {}) => {
const record: any = {
id,
...fields,
_preparedState: null,
prepareUpdate(recordUpdater: (m: any) => void) {
if (record._preparedState) {
throw new Error(`Cannot update a record with pending changes (${table}#${id})`);
}
recordUpdater(record);
record._preparedState = 'update';
return record;
}
};
return record;
};

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 makeMessageRecord = (id: string) => makeRecord('messages', id, { t: 'e2e', msg: 'cipher', subscription: { id: rid } });

const deferred = () => {
let resolve: () => void = () => undefined;
Expand Down Expand Up @@ -263,3 +263,39 @@ describe('Encryption.decryptPendingMessages', () => {
expect(failing.msg).toBe('cipher');
});
});

describe('Encryption.decryptPendingSubscriptions', () => {
const makeSubscriptionRecord = (id: string) =>
makeRecord('subscriptions', id, { lastMessage: { t: 'e2e', e2e: 'pending', msg: 'cipher' } });

beforeEach(() => {
jest.clearAllMocks();
mockQueryRows.subscriptions = [];
});

afterEach(() => {
jest.restoreAllMocks();
});

it('leaves no subscription prepared-but-uncommitted when one decryption rejects', async () => {
const healthy = makeSubscriptionRecord('s1');
const failing = makeSubscriptionRecord('s2');
mockQueryRows.subscriptions = [healthy, failing];
jest.spyOn(encryption, 'decryptSubscription').mockImplementation((sub: any) => {
if (sub.id === failing.id) {
return Promise.reject(new Error('decrypt failed'));
}
return Promise.resolve({ lastMessage: { t: 'e2e', e2e: 'done', msg: 'plain' } } as any);
});

await encryption.decryptPendingSubscriptions();

expect(healthy._preparedState).toBeNull();
expect(failing._preparedState).toBeNull();

const batched = mockDbBatch.mock.calls.flatMap(call => call.flat());
expect(batched).toContain(healthy);
expect(healthy.lastMessage).toEqual({ t: 'e2e', e2e: 'done', msg: 'plain' });
expect(failing.lastMessage).toEqual({ t: 'e2e', e2e: 'pending', msg: 'cipher' });
});
});
33 changes: 23 additions & 10 deletions app/lib/encryption/encryption.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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 {
Expand Down Expand Up @@ -64,6 +64,7 @@ import {
} from './utils';

const ROOM_KEY_EXCHANGE_SIZE = 10;

class Encryption {
ready: boolean;
privateKey: string | null;
Expand Down Expand Up @@ -379,7 +380,7 @@ class Encryption {
return null;
}
});
await db.batch(...prepared);
await db.batch(prepared.filter(record => record !== null));
});
} catch (e) {
log(e);
Expand All @@ -404,9 +405,23 @@ class Encryption {
sub => sub.lastMessage?.t === E2E_MESSAGE_TYPE && sub.lastMessage?.e2e !== E2E_STATUS.DONE
);

const preparedSubscriptions: (Model | null)[] = await Promise.all(
const decrypted = await Promise.all(
subsEncryptedToDecrypt.map(async (sub: TSubscriptionModel) => {
const newSub = await this.decryptSubscription(sub);
try {
return { sub, newSub: await this.decryptSubscription(sub) };
} catch (e) {
log(e);
return { sub, newSub: null };
}
})
);

if (!decrypted.length) {
return;
}

await db.write(async () => {
const prepared = decrypted.map(({ sub, newSub }) => {
try {
return sub.prepareUpdate(
protectedFunction((m: TSubscriptionModel) => {
Expand All @@ -415,14 +430,12 @@ class Encryption {
}
})
);
} catch {
} catch (e) {
log(e);
return null;
}
})
);

await db.write(async () => {
await db.batch(preparedSubscriptions.filter((record): record is Model => record !== null));
});
await db.batch(prepared.filter(record => record !== null));
});
} catch (e) {
log(e);
Expand Down
116 changes: 57 additions & 59 deletions app/lib/methods/subscriptions/rooms.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,6 +51,7 @@ const createOrUpdateSubscription = async (subscription: ISubscription, room: ISe
const db = database.active;
const subCollection = db.get('subscriptions');
const roomsCollection = db.get('rooms');
const messagesCollection = db.get('messages');

if (!subscription) {
try {
Expand Down Expand Up @@ -152,72 +153,69 @@ 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;
}
}
if (sub.hideUnreadStatus && subscription.hasOwnProperty('hideUnreadStatus')) {
if (sub.hideUnreadStatus !== subscription.hideUnreadStatus) {
s.hideUnreadStatus = !!subscription.hideUnreadStatus;
}
}
});
batch.push(update);
} catch (e) {
console.log(e);
}
} else {
const { subscribedRoom } = store.getState().room;
const lastMessage = tmp.lastMessage && subscribedRoom !== tmp.rid ? buildMessage(tmp.lastMessage) : null;
const messageRecord = lastMessage ? await getMessageById(lastMessage._id) : null;

await db.write(async () => {
const batch: Model[] = [];

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);
if (sub) {
batch.push(
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;
}
}
})
);
} else {
batch.push(
subCollection.prepareCreate(s => {
s._raw = sanitizedRaw({ id: tmp.rid }, subCollection.schema);
Object.assign(s, tmp);
if (s.roomUpdatedAt) {
s.roomUpdatedAt = new Date();
}
})
);
}
} catch (e) {
console.log(e);
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);
}

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);
})
);
try {
if (messageRecord) {
batch.push(
messageRecord.prepareUpdate(() => {
Object.assign(messageRecord, lastMessage);
})
);
} else {
batch.push(
messagesCollection.prepareCreate(m => {
m._raw = sanitizedRaw({ id: lastMessage._id }, messagesCollection.schema);
if (m.subscription) {
m.subscription.id = lastMessage.rid;
}
return Object.assign(m, lastMessage);
})
);
}
} catch (e) {
log(e);
}
}
}

await db.write(async () => {
await db.batch(batch);
});

Expand Down
Loading