From 078396e450c74c3d81e7593568c44267b352a866 Mon Sep 17 00:00:00 2001 From: Diego Mello Date: Tue, 4 Aug 2026 14:05:50 -0300 Subject: [PATCH 1/5] feat: expose a room-stream-ready signal on reconnect Emits a per-room event when the server acks the room's stream-room-messages subscription, on first connect and on every reconnect. No consumer yet. --- CONTEXT.md | 13 +- app/lib/methods/helpers/emitter.ts | 30 ++-- .../subscriptions/room.streamReady.test.ts | 149 ++++++++++++++++++ app/lib/methods/subscriptions/room.ts | 53 ++++++- app/lib/services/sdk.ts | 5 + 5 files changed, 230 insertions(+), 20 deletions(-) create mode 100644 app/lib/methods/subscriptions/room.streamReady.test.ts diff --git a/CONTEXT.md b/CONTEXT.md index 25a741e22f8..d6914f0d820 100644 --- a/CONTEXT.md +++ b/CONTEXT.md @@ -197,12 +197,13 @@ A **Message Action** is the active mode on a Message in the Room view. The three ## Server & Connection -| Term | Definition | Aliases to avoid | -| ------------------ | ------------------------------------------------------------------------------------------------------------------------ | ---------------------------------------------------------------- | -| **Server** | A Rocket.Chat server instance the app connects to, with version, settings, and enterprise modules | Workspace (used by web but not consistently in mobile), instance | -| **Server History** | List of previously connected Servers for quick reconnection | Recent servers | -| **Meteor Connect** | The WebSocket connection to the Server's DDP (Distributed Data Protocol) endpoint | Socket, connection | -| **Socket Health** | Whether the Meteor Connect socket is genuinely alive — confirmed by a round trip when in doubt, reopened when known dead | Staleness (stale/gray/fresh), socket probe | +| Term | Definition | Aliases to avoid | +| --------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ---------------------------------------------------------------- | +| **Server** | A Rocket.Chat server instance the app connects to, with version, settings, and enterprise modules | Workspace (used by web but not consistently in mobile), instance | +| **Server History** | List of previously connected Servers for quick reconnection | Recent servers | +| **Meteor Connect** | The WebSocket connection to the Server's DDP (Distributed Data Protocol) endpoint | Socket, connection | +| **Socket Health** | Whether the Meteor Connect socket is genuinely alive — confirmed by a round trip when in doubt, reopened when known dead | Staleness (stale/gray/fresh), socket probe | +| **Room Stream Ready** | The moment the Server acks a Room's `stream-room-messages` subscription, meaning live Messages for that Room are flowing again; fires on first connect and on every reconnect | Connected, subscribed | ## Navigation & Layout diff --git a/app/lib/methods/helpers/emitter.ts b/app/lib/methods/helpers/emitter.ts index 268ca8f0a96..fc1258c34a0 100644 --- a/app/lib/methods/helpers/emitter.ts +++ b/app/lib/methods/helpers/emitter.ts @@ -6,19 +6,27 @@ type TDynamicMediaDownloadEvents = { [key: `downloadMedia${string}`]: string; }; -export type TEmitterEvents = TDynamicMediaDownloadEvents & { - toolbarMention: undefined; - addMarkdown: { - style: TMarkdownStyle; - }; - setKeyboardHeight: number; - setKeyboardHeightThread: number; - setComposerHeight: number; - setComposerHeightThread: number; - audioFocused: string; - navigationReady: undefined; +/** Emitted once the server acks the room's `stream-room-messages` subscription, on every (re)connect. */ +type TRoomStreamReadyEvents = { + [key: `roomStreamReady${string}`]: undefined; }; +export const roomStreamReadyEvent = (rid: string) => `roomStreamReady${rid}` as const; + +export type TEmitterEvents = TDynamicMediaDownloadEvents & + TRoomStreamReadyEvents & { + toolbarMention: undefined; + addMarkdown: { + style: TMarkdownStyle; + }; + setKeyboardHeight: number; + setKeyboardHeightThread: number; + setComposerHeight: number; + setComposerHeightThread: number; + audioFocused: string; + navigationReady: undefined; + }; + export type TKeyEmitterEvent = keyof TEmitterEvents; export const emitter = mitt(); diff --git a/app/lib/methods/subscriptions/room.streamReady.test.ts b/app/lib/methods/subscriptions/room.streamReady.test.ts new file mode 100644 index 00000000000..4a953cc3c94 --- /dev/null +++ b/app/lib/methods/subscriptions/room.streamReady.test.ts @@ -0,0 +1,149 @@ +import RoomSubscription from './room'; +import sdk from '../../services/sdk'; +import { emitter, roomStreamReadyEvent } from '../helpers/emitter'; + +jest.mock('../../services/sdk', () => ({ + __esModule: true, + default: { + subscribeRoom: jest.fn(), + onStreamData: jest.fn(), + getSubscriptionById: jest.fn() + } +})); + +jest.mock('../../database', () => ({ + __esModule: true, + default: { active: { get: jest.fn(), write: jest.fn() } } +})); + +jest.mock('../../store/auxStore', () => ({ + store: { + getState: jest.fn(() => ({ server: { version: '7.4.0' }, settings: {}, login: { user: {} }, room: {} })), + dispatch: jest.fn() + } +})); + +jest.mock('../readMessages', () => ({ readMessages: jest.fn() })); +jest.mock('../loadMissedMessages', () => ({ loadMissedMessages: jest.fn() })); +jest.mock('../../encryption', () => ({ Encryption: { decryptMessage: jest.fn(m => m) } })); + +const mockedSubscribeRoom = sdk.subscribeRoom as jest.Mock; +const mockedOnStreamData = sdk.onStreamData as jest.Mock; +const mockedGetSubscriptionById = sdk.getSubscriptionById as jest.Mock; + +const RID = 'ROOM_ID'; +const MESSAGES_STREAM_ID = 'stream-room-messages-id'; + +const streamSubscriptions = () => [ + { id: MESSAGES_STREAM_ID, name: 'stream-room-messages', params: [RID], unsubscribe: jest.fn(() => Promise.resolve()) }, + { + id: 'notify-room-id', + name: 'stream-notify-room', + params: [`${RID}/typing`], + unsubscribe: jest.fn(() => Promise.resolve()) + } +]; + +describe('RoomSubscription stream ready signal', () => { + /** Listeners registered by the subscription, keyed by the DDP event they listen to. */ + let listeners: Record void>; + let onStreamReady: jest.Mock; + + beforeEach(() => { + jest.clearAllMocks(); + listeners = {}; + mockedOnStreamData.mockImplementation((event: string, callback: (message: any) => void) => { + listeners[event] = callback; + return Promise.resolve({ stop: jest.fn() }); + }); + mockedSubscribeRoom.mockResolvedValue(streamSubscriptions()); + // Mirrors the SDK: subscriptions are only registered once the server has acked them. + mockedGetSubscriptionById.mockImplementation((id: string) => streamSubscriptions().find(sub => sub.id === id)); + onStreamReady = jest.fn(); + emitter.on(roomStreamReadyEvent(RID), onStreamReady); + }); + + afterEach(() => { + emitter.off(roomStreamReadyEvent(RID), onStreamReady); + }); + + it('fires once the server acks the room messages stream on first connect', async () => { + await new RoomSubscription(RID).subscribe(); + + expect(onStreamReady).toHaveBeenCalledTimes(1); + }); + + it('does not fire at socket open, only when the ack arrives', async () => { + let ackSubscriptions: (subscriptions: unknown[]) => void = () => {}; + mockedSubscribeRoom.mockReturnValue( + new Promise(resolve => { + ackSubscriptions = resolve; + }) + ); + + await new RoomSubscription(RID).subscribe(); + // socket is open and the listeners are wired, but the server hasn't acked yet + listeners.connected?.({}); + await Promise.resolve(); + expect(onStreamReady).not.toHaveBeenCalled(); + + ackSubscriptions(streamSubscriptions()); + await Promise.resolve(); + await Promise.resolve(); + + expect(onStreamReady).toHaveBeenCalledTimes(1); + }); + + it('fires again on every reconnect that re-acks the same subscription id', async () => { + await new RoomSubscription(RID).subscribe(); + onStreamReady.mockClear(); + + listeners.ready({ msg: 'ready', subs: [MESSAGES_STREAM_ID] }); + listeners.ready({ msg: 'ready', subs: [MESSAGES_STREAM_ID] }); + + expect(onStreamReady).toHaveBeenCalledTimes(2); + }); + + it('still fires on reconnect when the first connect never acked', async () => { + mockedSubscribeRoom.mockRejectedValue(new Error('socket closed')); + + await new RoomSubscription(RID).subscribe(); + await Promise.resolve(); + expect(onStreamReady).not.toHaveBeenCalled(); + + listeners.ready({ msg: 'ready', subs: [MESSAGES_STREAM_ID] }); + + expect(onStreamReady).toHaveBeenCalledTimes(1); + }); + + it('ignores an ack for another room messages stream', async () => { + await new RoomSubscription(RID).subscribe(); + onStreamReady.mockClear(); + mockedGetSubscriptionById.mockReturnValue({ id: 'other', name: 'stream-room-messages', params: ['OTHER_ROOM'] }); + + listeners.ready({ msg: 'ready', subs: ['other'] }); + + expect(onStreamReady).not.toHaveBeenCalled(); + }); + + it('ignores acks for other subscriptions', async () => { + await new RoomSubscription(RID).subscribe(); + onStreamReady.mockClear(); + + listeners.ready({ msg: 'ready', subs: ['some-other-subscription'] }); + + expect(onStreamReady).not.toHaveBeenCalled(); + }); + + it('stops firing after unsubscribe', async () => { + const subscription = new RoomSubscription(RID); + await subscription.subscribe(); + const handleStreamReady = listeners.ready; + await subscription.unsubscribe(); + onStreamReady.mockClear(); + + handleStreamReady({ msg: 'ready', subs: [MESSAGES_STREAM_ID] }); + + expect(onStreamReady).not.toHaveBeenCalled(); + }); +}); diff --git a/app/lib/methods/subscriptions/room.ts b/app/lib/methods/subscriptions/room.ts index acfe1c74029..8b374f84cbd 100644 --- a/app/lib/methods/subscriptions/room.ts +++ b/app/lib/methods/subscriptions/room.ts @@ -18,25 +18,41 @@ import { Encryption } from '../../encryption'; import { type IMessage, type TMessageModel, - type TSubscriptionModel, type TThreadMessageModel, type TThreadModel, type IDeleteMessageBulkParams } from '../../../definitions'; +import { emitter, roomStreamReadyEvent } from '../helpers/emitter'; import { type IDDPMessage } from '../../../definitions/IDDPMessage'; import sdk from '../../services/sdk'; import { readMessages } from '../readMessages'; import { loadMissedMessages } from '../loadMissedMessages'; import markMessagesRead from '../helpers/markMessagesRead'; +/** A DDP stream subscription as returned by the SDK — not a WatermelonDB subscription record. */ +interface IStreamSubscription { + id: string; + name: string; + params: any[]; + unsubscribe: () => Promise; +} + +const MESSAGES_STREAM = 'stream-room-messages'; + +/** DDP `ready` message: the server acking one or more subscription ids. */ +interface IDDPReadyMessage { + subs?: string[]; +} + export default class RoomSubscription { private rid: string; private isAlive: boolean; - private promises?: Promise; + private promises?: Promise<(IStreamSubscription | undefined)[]>; private connectedListener?: Promise; private disconnectedListener?: Promise; private notifyRoomListener?: Promise; private messageReceivedListener?: Promise; + private streamReadyListener?: Promise; constructor(rid: string) { this.rid = rid; @@ -49,7 +65,20 @@ export default class RoomSubscription { await this.unsubscribe(); } this.promises = sdk.subscribeRoom(this.rid); + // The `sub` request resolves on the server's `ready` ack, and the subscription is only + // registered on the SDK afterwards — so the first connect is signalled from here, and + // every later re-ack from `handleStreamReady`. + this.promises + .then(subscriptions => { + if (this.isAlive && subscriptions?.some(subscription => subscription?.name === MESSAGES_STREAM)) { + emitter.emit(roomStreamReadyEvent(this.rid)); + } + }) + .catch(() => { + // do nothing + }); + this.streamReadyListener = sdk.onStreamData('ready', this.handleStreamReady); this.connectedListener = sdk.onStreamData('connected', this.handleConnection); this.disconnectedListener = sdk.onStreamData('close', this.handleConnection); this.notifyRoomListener = sdk.onStreamData('stream-notify-room', this.handleNotifyRoomReceived); @@ -68,7 +97,7 @@ export default class RoomSubscription { if (this.promises) { try { const subscriptions = (await this.promises) || []; - subscriptions.forEach(sub => sub.unsubscribe().catch(() => console.log('unsubscribeRoom'))); + subscriptions.forEach(sub => sub?.unsubscribe().catch(() => console.log('unsubscribeRoom'))); } catch (e) { // do nothing } @@ -78,6 +107,24 @@ export default class RoomSubscription { this.removeListener(this.disconnectedListener); this.removeListener(this.notifyRoomListener); this.removeListener(this.messageReceivedListener); + this.removeListener(this.streamReadyListener); + }; + + /** + * The subscription is re-sent with its original id after each reconnect, so the acked ids are + * resolved against the SDK's live subscriptions instead of an id captured on the first connect. + */ + handleStreamReady = (ddpMessage: IDDPReadyMessage) => { + if (!this.isAlive) { + return; + } + const isMessagesStream = ddpMessage?.subs?.some(id => { + const subscription = sdk.getSubscriptionById(id); + return subscription?.name === MESSAGES_STREAM && subscription?.params?.[0] === this.rid; + }); + if (isMessagesStream) { + emitter.emit(roomStreamReadyEvent(this.rid)); + } }; removeListener = async (promise?: Promise): Promise => { diff --git a/app/lib/services/sdk.ts b/app/lib/services/sdk.ts index d8be776f45a..6684ce7b667 100644 --- a/app/lib/services/sdk.ts +++ b/app/lib/services/sdk.ts @@ -181,6 +181,11 @@ class Sdk { ]); } + /** Look up a live DDP subscription by the id the server acks in a `ready` message. */ + getSubscriptionById(id: string) { + return this.current?.ddp?.subscriptions?.[id]; + } + unsubscribe(subscription: any[]) { return this.current.unsubscribe(subscription); } From 1faa8a4de01effc66eab378e2e7e5ec3a3cdf738 Mon Sep 17 00:00:00 2001 From: Diego Mello Date: Tue, 4 Aug 2026 15:15:05 -0300 Subject: [PATCH 2/5] fix: run the room missed-messages fetch after the room stream is live MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The fetch ran on the SDK `connected` event — the raw socket open, before the DDP handshake and before login — while the room's stream-room-messages subscription is only re-sent after the resume login. Messages the server accepted in between reached neither the fetch nor the stream, and nothing fetched again, so an open room silently lost them until a full app restart. The fetch now runs on the room stream ready signal, so its snapshot overlaps the live stream. It only runs when the socket dropped or re-handshaked since the last fetch, so opening a room still leaves the initial load to RoomView, and it keeps that flag set until a fetch succeeds so a failed one is retried on the next ack. How the `lastOpen` cursor is computed is untouched. --- .../subscriptions/room.reconnectFetch.test.ts | 258 ++++++++++++++++++ .../subscriptions/room.resumeSync.test.ts | 4 +- app/lib/methods/subscriptions/room.ts | 41 ++- .../subscriptions/roomCloseCursor.test.ts | 2 +- 4 files changed, 296 insertions(+), 9 deletions(-) create mode 100644 app/lib/methods/subscriptions/room.reconnectFetch.test.ts diff --git a/app/lib/methods/subscriptions/room.reconnectFetch.test.ts b/app/lib/methods/subscriptions/room.reconnectFetch.test.ts new file mode 100644 index 00000000000..b733e7d6782 --- /dev/null +++ b/app/lib/methods/subscriptions/room.reconnectFetch.test.ts @@ -0,0 +1,258 @@ +import EJSON from 'ejson'; + +import RoomSubscription from './room'; +import sdk from '../../services/sdk'; +import updateMessages from '../updateMessages'; +import { getSubscriptionByRoomId } from '../../database/services/Subscription'; + +jest.mock('../../services/sdk', () => ({ + __esModule: true, + default: { + get: jest.fn(), + subscribeRoom: jest.fn(), + onStreamData: jest.fn(), + getSubscriptionById: jest.fn() + } +})); + +const batched: any[] = []; +const messagesCollection = { + schema: { columns: {}, columnArray: [] }, + prepareCreate: (build: (record: any) => void) => { + const record: any = {}; + build(record); + return record; + } +}; + +jest.mock('../../database', () => ({ + __esModule: true, + default: { + active: { + get: jest.fn(() => messagesCollection), + write: jest.fn((work: () => Promise) => work()), + batch: jest.fn((...records: any[]) => { + batched.push(...records.filter(Boolean)); + }) + } + } +})); + +jest.mock('../../database/services/Subscription', () => ({ + getSubscriptionByRoomId: jest.fn() +})); + +jest.mock('../../database/services/Message', () => ({ getMessageById: jest.fn(() => Promise.resolve(null)) })); +jest.mock('../../database/services/Thread', () => ({ getThreadById: jest.fn(() => Promise.resolve(null)) })); +jest.mock('../../database/services/ThreadMessage', () => ({ getThreadMessageById: jest.fn(() => Promise.resolve(null)) })); + +jest.mock('../../store/auxStore', () => ({ + store: { + getState: jest.fn(() => ({ server: { version: '7.4.0' }, settings: {}, login: { user: {} }, room: {} })), + dispatch: jest.fn() + } +})); + +jest.mock('../updateMessages', () => jest.fn()); +jest.mock('../readMessages', () => ({ readMessages: jest.fn() })); +jest.mock('../../encryption', () => ({ Encryption: { decryptMessage: jest.fn(m => m) } })); + +const mockedSdkGet = sdk.get as jest.MockedFunction; +const mockedSubscribeRoom = sdk.subscribeRoom as jest.Mock; +const mockedOnStreamData = sdk.onStreamData as jest.Mock; +const mockedGetSubscriptionById = sdk.getSubscriptionById as jest.Mock; +const mockedUpdateMessages = updateMessages as jest.MockedFunction; +const mockedGetSubscriptionByRoomId = getSubscriptionByRoomId as jest.MockedFunction; + +const RID = 'ROOM_ID'; +const MESSAGES_STREAM_ID = 'stream-room-messages-id'; +const CURSOR = new Date(Date.UTC(2024, 0, 1, 11, 0, 0)); + +/** Sent while the socket was down: only the catch-up fetch can bring it in. */ +const offlineMessage = { + _id: 'offline-1', + rid: RID, + msg: 'sent while the socket was down', + ts: new Date(Date.UTC(2024, 0, 1, 11, 30, 0)).toISOString(), + _updatedAt: new Date(Date.UTC(2024, 0, 1, 11, 30, 0)).toISOString(), + u: { _id: 'user2', username: 'user2' } +}; + +/** Accepted by the server between the socket opening and the stream being acked — the lost window. */ +const windowMessage = { + _id: 'window-1', + rid: RID, + msg: 'sent while the room stream was still subscribing', + ts: new Date(Date.UTC(2024, 0, 1, 11, 59, 0)).toISOString(), + _updatedAt: new Date(Date.UTC(2024, 0, 1, 11, 59, 0)).toISOString(), + u: { _id: 'user2', username: 'user2' } +}; + +/** Delivered by the live stream once it is acked. */ +const streamedMessage = { + _id: 'streamed-1', + rid: RID, + msg: 'sent after the room stream was acked', + ts: { $date: Date.UTC(2024, 0, 1, 12, 0, 0) }, + u: { _id: 'user2', username: 'user2' } +}; + +const streamSubscriptions = () => [ + { id: MESSAGES_STREAM_ID, name: 'stream-room-messages', params: [RID], unsubscribe: jest.fn(() => Promise.resolve()) } +]; + +describe('RoomSubscription reconnect catch-up fetch', () => { + /** Listeners registered by the subscription, keyed by the DDP event they listen to. */ + let listeners: Record void>; + + /** The socket dropping and reopening, up to the DDP handshake — all before the stream ack. */ + const reconnectSocket = () => { + listeners.close?.({}); + listeners.connected?.({}); + }; + + const ackRoomStream = () => listeners.ready({ msg: 'ready', subs: [MESSAGES_STREAM_ID] }); + + const flush = () => new Promise(resolve => setImmediate(resolve)); + + /** One fetch issues an UPDATED and a DELETED request; the UPDATED ones count the fetches. */ + const fetchCount = () => + mockedSdkGet.mock.calls.filter(([endpoint, params]: any[]) => endpoint === 'chat.syncMessages' && params?.type === 'UPDATED') + .length; + + beforeEach(() => { + jest.clearAllMocks(); + batched.length = 0; + listeners = {}; + mockedOnStreamData.mockImplementation((event: string, callback: (message: any) => void) => { + listeners[event] = callback; + return Promise.resolve({ stop: jest.fn() }); + }); + mockedSubscribeRoom.mockResolvedValue(streamSubscriptions()); + mockedGetSubscriptionById.mockImplementation((id: string) => streamSubscriptions().find(sub => sub.id === id)); + mockedUpdateMessages.mockResolvedValue(0); + mockedGetSubscriptionByRoomId.mockResolvedValue({ lastOpen: CURSOR } as never); + mockedSdkGet.mockResolvedValue({ + result: { updated: [offlineMessage, windowMessage], deleted: [], cursor: { next: null } } + } as never); + }); + + const openRoom = async () => { + const subscription = new RoomSubscription(RID); + await subscription.subscribe(); + await flush(); + // opening the room acks the stream too; RoomView owns that load, so nothing is fetched here + mockedSdkGet.mockClear(); + mockedUpdateMessages.mockClear(); + return subscription; + }; + + it('does not fetch while the room stream is still subscribing', async () => { + await openRoom(); + + reconnectSocket(); + await flush(); + + expect(mockedSdkGet).not.toHaveBeenCalled(); + }); + + it('fetches once the room stream is acked', async () => { + await openRoom(); + + reconnectSocket(); + ackRoomStream(); + await flush(); + + expect(mockedSdkGet).toHaveBeenCalledWith( + 'chat.syncMessages', + expect.objectContaining({ roomId: RID, type: 'UPDATED', next: CURSOR.getTime() }) + ); + }); + + it('persists a message set straddling the reconnect window', async () => { + await openRoom(); + + reconnectSocket(); + ackRoomStream(); + await flush(); + await listeners['stream-room-messages']({ fields: { args: [EJSON.toJSONValue(streamedMessage)] } }); + + expect(mockedUpdateMessages).toHaveBeenCalledWith( + expect.objectContaining({ + rid: RID, + update: expect.arrayContaining([ + expect.objectContaining({ _id: offlineMessage._id }), + expect.objectContaining({ _id: windowMessage._id }) + ]) + }) + ); + expect(batched).toEqual(expect.arrayContaining([expect.objectContaining({ _id: streamedMessage._id })])); + }); + + it('does not fetch when the room is opened on a healthy socket', async () => { + const subscription = new RoomSubscription(RID); + await subscription.subscribe(); + await flush(); + ackRoomStream(); + await flush(); + + expect(mockedSdkGet).not.toHaveBeenCalled(); + }); + + it('fetches nothing for a room without a sync cursor: RoomView owns the initial load', async () => { + await openRoom(); + mockedGetSubscriptionByRoomId.mockResolvedValue({ lastOpen: null, t: 'c' } as never); + + reconnectSocket(); + ackRoomStream(); + await flush(); + + expect(mockedSdkGet).not.toHaveBeenCalled(); + }); + + it('retries on the next ack when the fetch fails', async () => { + await openRoom(); + mockedSdkGet.mockRejectedValueOnce(new Error('socket closed mid-fetch')); + + reconnectSocket(); + ackRoomStream(); + await flush(); + ackRoomStream(); + await flush(); + + expect(fetchCount()).toBe(2); + }); + + it('fetches after a socket reopen that emits no close', async () => { + await openRoom(); + + listeners.connected({}); + ackRoomStream(); + await flush(); + + expect(fetchCount()).toBe(1); + }); + + it('does not fetch again on an ack with no reconnect in between', async () => { + await openRoom(); + + reconnectSocket(); + ackRoomStream(); + await flush(); + ackRoomStream(); + await flush(); + + expect(fetchCount()).toBe(1); + }); + + it('does not fetch after the room is left', async () => { + const subscription = await openRoom(); + reconnectSocket(); + await subscription.unsubscribe(); + + ackRoomStream(); + await flush(); + + expect(mockedSdkGet).not.toHaveBeenCalled(); + }); +}); diff --git a/app/lib/methods/subscriptions/room.resumeSync.test.ts b/app/lib/methods/subscriptions/room.resumeSync.test.ts index 6d5004403f5..5e70e9e0412 100644 --- a/app/lib/methods/subscriptions/room.resumeSync.test.ts +++ b/app/lib/methods/subscriptions/room.resumeSync.test.ts @@ -65,7 +65,7 @@ describe('RoomSubscription resume sync', () => { const persistedCursor = new Date(Date.UTC(2024, 0, 1, 11, 0, 0)); mockedGetSubscriptionByRoomId.mockResolvedValue({ lastOpen: persistedCursor } as never); - await new RoomSubscription(RID).handleConnection(); + await new RoomSubscription(RID).fetchMissedMessages(); expect(mockedSdkGet).toHaveBeenCalledWith( 'chat.syncMessages', @@ -82,7 +82,7 @@ describe('RoomSubscription resume sync', () => { it('fetches nothing for a room without a sync cursor (null lastOpen): RoomView owns the initial load', async () => { mockedGetSubscriptionByRoomId.mockResolvedValue({ lastOpen: null, t: 'c' } as never); - await new RoomSubscription(RID).handleConnection(); + await new RoomSubscription(RID).fetchMissedMessages(); expect(mockedSdkGet).not.toHaveBeenCalled(); }); diff --git a/app/lib/methods/subscriptions/room.ts b/app/lib/methods/subscriptions/room.ts index 8b374f84cbd..6114be4e04c 100644 --- a/app/lib/methods/subscriptions/room.ts +++ b/app/lib/methods/subscriptions/room.ts @@ -53,10 +53,13 @@ export default class RoomSubscription { private notifyRoomListener?: Promise; private messageReceivedListener?: Promise; private streamReadyListener?: Promise; + /** Set on socket close and on a new DDP handshake, so opening a room isn't mistaken for a reconnect. */ + private hasReconnected: boolean; constructor(rid: string) { this.rid = rid; this.isAlive = true; + this.hasReconnected = false; } subscribe = async () => { @@ -78,9 +81,12 @@ export default class RoomSubscription { // do nothing }); + // The catch-up fetch runs on the room stream ready signal, not on the raw socket open: + // its snapshot has to overlap the live stream, or messages accepted in between are lost. + emitter.on(roomStreamReadyEvent(this.rid), this.handleStreamReadySignal); this.streamReadyListener = sdk.onStreamData('ready', this.handleStreamReady); - this.connectedListener = sdk.onStreamData('connected', this.handleConnection); - this.disconnectedListener = sdk.onStreamData('close', this.handleConnection); + this.connectedListener = sdk.onStreamData('connected', this.handleReconnection); + this.disconnectedListener = sdk.onStreamData('close', this.handleDisconnection); this.notifyRoomListener = sdk.onStreamData('stream-notify-room', this.handleNotifyRoomReceived); this.messageReceivedListener = sdk.onStreamData('stream-room-messages', this.handleMessageReceived); if (!this.isAlive) { @@ -103,6 +109,7 @@ export default class RoomSubscription { } } reduxStore.dispatch(clearUserTyping()); + emitter.off(roomStreamReadyEvent(this.rid), this.handleStreamReadySignal); this.removeListener(this.connectedListener); this.removeListener(this.disconnectedListener); this.removeListener(this.notifyRoomListener); @@ -138,16 +145,38 @@ export default class RoomSubscription { } }; - handleConnection = async () => { + handleDisconnection = () => { + this.hasReconnected = true; + reduxStore.dispatch(clearUserTyping()); + }; + + /** A new DDP handshake also means a reconnect, including reopens that emit no `close`. */ + handleReconnection = () => { + this.hasReconnected = true; + }; + + /** + * Fetches the missed messages once the room's stream is live again, so the fetch snapshot and + * the stream overlap. Opening a room also acks the stream, and RoomView owns that initial load. + */ + handleStreamReadySignal = async () => { + if (!this.isAlive || !this.hasReconnected) { + return; + } try { - reduxStore.dispatch(clearUserTyping()); - await loadMissedMessages({ rid: this.rid }); - this.read(); + await this.fetchMissedMessages(); + // Kept set until the fetch succeeds, so a failed one is retried on the next ack. + this.hasReconnected = false; } catch (e) { log(e); } }; + fetchMissedMessages = async () => { + await loadMissedMessages({ rid: this.rid }); + this.read(); + }; + handleNotifyRoomReceived = protectedFunction(async (ddpMessage: IDDPMessage) => { const [_rid, ev] = ddpMessage.fields.eventName.split('/'); if (this.rid !== _rid) { diff --git a/app/lib/methods/subscriptions/roomCloseCursor.test.ts b/app/lib/methods/subscriptions/roomCloseCursor.test.ts index 62e3892c8fa..fd1d9f3a52a 100644 --- a/app/lib/methods/subscriptions/roomCloseCursor.test.ts +++ b/app/lib/methods/subscriptions/roomCloseCursor.test.ts @@ -115,7 +115,7 @@ describe('closing a room while offline must not advance the sync cursor', () => await new RoomSubscription(RID).unsubscribe(); await Promise.resolve(); - await new RoomSubscription(RID).handleConnection(); + await new RoomSubscription(RID).fetchMissedMessages(); expect(mockedUpdateMessages).toHaveBeenCalledWith( expect.objectContaining({ From 8be135dfcaa607b792a13fe1862dc7d6bc5eb420 Mon Sep 17 00:00:00 2001 From: Diego Mello Date: Tue, 4 Aug 2026 15:31:03 -0300 Subject: [PATCH 3/5] fix: drop stale catch-up fetches across connection cycles MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A catch-up fetch started around a socket close can resolve long after a new connection cycle already fetched. Its pages are older, and the sync cursor has no monotonic clamp, so a late result could overwrite rows the live stream had already updated and lower `lastOpen`. Each fetch now captures the connection cycle it belongs to and is dropped — before its pages are written, before it paginates further, and before the cursor is touched — once that cycle has ended or the room was left. Dropping it does not clear the reconnect flag, so the current cycle still fetches on its own ack. How the cursor is computed is untouched. --- app/lib/methods/loadMissedMessages.ts | 17 +- .../subscriptions/room.resumeSync.test.ts | 4 +- .../subscriptions/room.staleFetch.test.ts | 231 ++++++++++++++++++ app/lib/methods/subscriptions/room.ts | 23 +- .../subscriptions/roomCloseCursor.test.ts | 2 +- 5 files changed, 270 insertions(+), 7 deletions(-) create mode 100644 app/lib/methods/subscriptions/room.staleFetch.test.ts diff --git a/app/lib/methods/loadMissedMessages.ts b/app/lib/methods/loadMissedMessages.ts index be285d1eabf..a77920c9cc8 100644 --- a/app/lib/methods/loadMissedMessages.ts +++ b/app/lib/methods/loadMissedMessages.ts @@ -76,6 +76,11 @@ export async function loadMissedMessages(args: { updatedNext?: number | null; deletedNext?: number | null; serverTimestamps?: TServerTimestamps; + /** + * Checked after every page: a fetch belonging to a connection cycle that has already ended can + * resolve late and overwrite newer rows or lower the cursor, so its result is dropped instead. + */ + isStale?: () => boolean; }): Promise { // A DELETED-only continuation fetches no UPDATED page, so it must not write the cursor again. const fetchedUpdatedPage = !!args.updatedNext || !args.deletedNext; @@ -84,6 +89,9 @@ export async function loadMissedMessages(args: { updatedNext: args.updatedNext, deletedNext: args.deletedNext }); + if (args.isStale?.()) { + return; + } if (data) { const { updated, @@ -97,12 +105,19 @@ export async function loadMissedMessages(args: { // @ts-ignore // TODO: remove loaderItem obligatoriness await updateMessages({ rid: args.rid, update: updated, remove: deleted }); + // Re-checked because the write above awaits: the cycle can end while it runs, and the cursor + // has no monotonic clamp, so a stale write would lower it. + if (args.isStale?.()) { + return; + } + if (deletedNext || updatedNext) { loadMissedMessages({ rid: args.rid, updatedNext, deletedNext, - serverTimestamps + serverTimestamps, + isStale: args.isStale }).catch(log); } diff --git a/app/lib/methods/subscriptions/room.resumeSync.test.ts b/app/lib/methods/subscriptions/room.resumeSync.test.ts index 5e70e9e0412..f4525b95409 100644 --- a/app/lib/methods/subscriptions/room.resumeSync.test.ts +++ b/app/lib/methods/subscriptions/room.resumeSync.test.ts @@ -65,7 +65,7 @@ describe('RoomSubscription resume sync', () => { const persistedCursor = new Date(Date.UTC(2024, 0, 1, 11, 0, 0)); mockedGetSubscriptionByRoomId.mockResolvedValue({ lastOpen: persistedCursor } as never); - await new RoomSubscription(RID).fetchMissedMessages(); + await new RoomSubscription(RID).fetchMissedMessages(() => false); expect(mockedSdkGet).toHaveBeenCalledWith( 'chat.syncMessages', @@ -82,7 +82,7 @@ describe('RoomSubscription resume sync', () => { it('fetches nothing for a room without a sync cursor (null lastOpen): RoomView owns the initial load', async () => { mockedGetSubscriptionByRoomId.mockResolvedValue({ lastOpen: null, t: 'c' } as never); - await new RoomSubscription(RID).fetchMissedMessages(); + await new RoomSubscription(RID).fetchMissedMessages(() => false); expect(mockedSdkGet).not.toHaveBeenCalled(); }); diff --git a/app/lib/methods/subscriptions/room.staleFetch.test.ts b/app/lib/methods/subscriptions/room.staleFetch.test.ts new file mode 100644 index 00000000000..3ab76cbf9b1 --- /dev/null +++ b/app/lib/methods/subscriptions/room.staleFetch.test.ts @@ -0,0 +1,231 @@ +import RoomSubscription from './room'; +import sdk from '../../services/sdk'; +import updateMessages from '../updateMessages'; +import { getSubscriptionByRoomId } from '../../database/services/Subscription'; +import { updateLastOpen } from '../updateLastOpen'; +import { readMessages } from '../readMessages'; + +jest.mock('../../services/sdk', () => ({ + __esModule: true, + default: { + get: jest.fn(), + subscribeRoom: jest.fn(), + onStreamData: jest.fn(), + getSubscriptionById: jest.fn() + } +})); + +jest.mock('../../database', () => ({ + __esModule: true, + default: { + active: { + get: jest.fn(() => ({ schema: { columns: {}, columnArray: [] }, prepareCreate: jest.fn() })), + write: jest.fn((work: () => Promise) => work()), + batch: jest.fn() + } + } +})); + +jest.mock('../../database/services/Subscription', () => ({ + getSubscriptionByRoomId: jest.fn() +})); + +jest.mock('../../store/auxStore', () => ({ + store: { + getState: jest.fn(() => ({ server: { version: '7.4.0' }, settings: {}, login: { user: {} }, room: {} })), + dispatch: jest.fn() + } +})); + +jest.mock('../updateMessages', () => jest.fn()); +jest.mock('../updateLastOpen', () => ({ + updateLastOpen: jest.fn(), + snapshotServerTimestamps: jest.requireActual('../updateLastOpen').snapshotServerTimestamps +})); +jest.mock('../readMessages', () => ({ readMessages: jest.fn() })); +jest.mock('../../encryption', () => ({ Encryption: { decryptMessage: jest.fn(m => m) } })); + +const mockedSdkGet = sdk.get as jest.MockedFunction; +const mockedSubscribeRoom = sdk.subscribeRoom as jest.Mock; +const mockedOnStreamData = sdk.onStreamData as jest.Mock; +const mockedGetSubscriptionById = sdk.getSubscriptionById as jest.Mock; +const mockedUpdateMessages = updateMessages as jest.MockedFunction; +const mockedUpdateLastOpen = updateLastOpen as jest.MockedFunction; +const mockedGetSubscriptionByRoomId = getSubscriptionByRoomId as jest.MockedFunction; + +const RID = 'ROOM_ID'; +const MESSAGES_STREAM_ID = 'stream-room-messages-id'; +const CURSOR = new Date(Date.UTC(2024, 0, 1, 11, 0, 0)); + +const message = (id: string, minute: number) => ({ + _id: id, + rid: RID, + msg: id, + ts: new Date(Date.UTC(2024, 0, 1, 11, minute, 0)).toISOString(), + _updatedAt: new Date(Date.UTC(2024, 0, 1, 11, minute, 0)).toISOString(), + u: { _id: 'user2', username: 'user2' } +}); + +/** From the connection cycle that already ended: its result must not land. */ +const staleMessage = message('stale-1', 10); +/** From the current connection cycle. */ +const freshMessage = message('fresh-1', 40); + +const streamSubscriptions = () => [ + { id: MESSAGES_STREAM_ID, name: 'stream-room-messages', params: [RID], unsubscribe: jest.fn(() => Promise.resolve()) } +]; + +describe('RoomSubscription stale catch-up fetch', () => { + let listeners: Record void>; + /** Resolvers for every pending `chat.syncMessages` request, in call order. */ + let pending: ((result: any) => void)[]; + /** Subscriptions opened by a test, torn down afterwards. */ + let opened: RoomSubscription[]; + + const reconnectSocket = () => { + listeners.close?.({}); + listeners.connected?.({}); + }; + + const ackRoomStream = () => listeners.ready({ msg: 'ready', subs: [MESSAGES_STREAM_ID] }); + + const flush = () => new Promise(resolve => setImmediate(resolve)); + + /** Resolves the n-th (0-based) `chat.syncMessages` request, in call order. */ + const resolveRequest = async (index: number, result: any) => { + pending[index]?.({ result }); + await flush(); + }; + + /** + * Resolves both requests of the n-th (0-based) fetch with `updated` and no further page. Only + * valid while every fetch issues an UPDATED and a DELETED request — a continuation may issue one. + */ + const resolveFetch = async (index: number, updated: any[], next: number | null = null) => { + await resolveRequest(index * 2, { updated, deleted: [], cursor: { next } }); + await resolveRequest(index * 2 + 1, { deleted: [], cursor: { next: null } }); + }; + + beforeEach(() => { + jest.clearAllMocks(); + listeners = {}; + pending = []; + opened = []; + mockedOnStreamData.mockImplementation((event: string, callback: (message: any) => void) => { + listeners[event] = callback; + return Promise.resolve({ stop: jest.fn() }); + }); + mockedSubscribeRoom.mockResolvedValue(streamSubscriptions()); + mockedGetSubscriptionById.mockImplementation((id: string) => streamSubscriptions().find(sub => sub.id === id)); + mockedUpdateMessages.mockResolvedValue(0); + mockedGetSubscriptionByRoomId.mockResolvedValue({ lastOpen: CURSOR } as never); + mockedSdkGet.mockImplementation(() => new Promise(resolve => pending.push(resolve)) as never); + }); + + // The room stream ready signal goes through a module-level emitter, so a subscription left alive + // keeps fetching into the next test. + afterEach(async () => { + await Promise.all(opened.map(subscription => subscription.unsubscribe())); + }); + + const openRoom = async () => { + const subscription = new RoomSubscription(RID); + opened.push(subscription); + await subscription.subscribe(); + await flush(); + return subscription; + }; + + it('ignores a fetch from a previous connection cycle resolving after the current one', async () => { + await openRoom(); + + // cycle 1: fetch starts and stays in flight + reconnectSocket(); + ackRoomStream(); + await flush(); + + // cycle 2: a second reconnect fetches and lands first + reconnectSocket(); + ackRoomStream(); + await flush(); + await resolveFetch(1, [freshMessage]); + + // cycle 1's fetch resolves late + await resolveFetch(0, [staleMessage]); + + expect(mockedUpdateMessages).toHaveBeenCalledTimes(1); + expect(mockedUpdateMessages).toHaveBeenCalledWith( + expect.objectContaining({ update: [expect.objectContaining({ _id: freshMessage._id })] }) + ); + }); + + it('does not move the sync cursor from a previous connection cycle', async () => { + await openRoom(); + + reconnectSocket(); + ackRoomStream(); + await flush(); + + reconnectSocket(); + ackRoomStream(); + await flush(); + await resolveFetch(1, [freshMessage]); + await resolveFetch(0, [staleMessage]); + + expect(mockedUpdateLastOpen).toHaveBeenCalledTimes(1); + expect(mockedUpdateLastOpen).toHaveBeenCalledWith(RID, [expect.objectContaining({ _updatedAt: freshMessage._updatedAt })]); + }); + + it('ignores a fetch that resolves after the room is left', async () => { + const subscription = await openRoom(); + + reconnectSocket(); + ackRoomStream(); + await flush(); + await subscription.unsubscribe(); + await resolveFetch(0, [staleMessage]); + + expect(mockedUpdateMessages).not.toHaveBeenCalled(); + expect(mockedUpdateLastOpen).not.toHaveBeenCalled(); + expect(readMessages).not.toHaveBeenCalled(); + }); + + it('stops paginating when the connection cycle ends mid-walk', async () => { + await openRoom(); + + reconnectSocket(); + ackRoomStream(); + await flush(); + // first page lands while the cycle is still current, and asks for another + await resolveFetch(0, [freshMessage], new Date(freshMessage._updatedAt).getTime()); + + reconnectSocket(); + // the continuation resolves after the cycle ended: no write, and the cursor stays put + await resolveRequest(2, { updated: [staleMessage], deleted: [], cursor: { next: null } }); + + expect(mockedUpdateMessages).toHaveBeenCalledTimes(1); + expect(mockedUpdateLastOpen).not.toHaveBeenCalled(); + }); + + it('still fetches on the next ack after a stale fetch was dropped', async () => { + await openRoom(); + + reconnectSocket(); + ackRoomStream(); + await flush(); + + // the socket drops again before the fetch resolves, so its result is dropped — and dropping + // it must not mark the reconnect as caught up + reconnectSocket(); + await resolveFetch(0, [staleMessage]); + + ackRoomStream(); + await flush(); + await resolveFetch(1, [freshMessage]); + + expect(mockedUpdateMessages).toHaveBeenCalledTimes(1); + expect(mockedUpdateMessages).toHaveBeenCalledWith( + expect.objectContaining({ update: [expect.objectContaining({ _id: freshMessage._id })] }) + ); + }); +}); diff --git a/app/lib/methods/subscriptions/room.ts b/app/lib/methods/subscriptions/room.ts index 6114be4e04c..31acc4ffe96 100644 --- a/app/lib/methods/subscriptions/room.ts +++ b/app/lib/methods/subscriptions/room.ts @@ -55,11 +55,17 @@ export default class RoomSubscription { private streamReadyListener?: Promise; /** Set on socket close and on a new DDP handshake, so opening a room isn't mistaken for a reconnect. */ private hasReconnected: boolean; + /** + * Bumped on both socket close and DDP handshake, so a fetch can tell its cycle has ended — only + * compared for equality, it is not a count of reconnects. + */ + private connectionCycle: number; constructor(rid: string) { this.rid = rid; this.isAlive = true; this.hasReconnected = false; + this.connectionCycle = 0; } subscribe = async () => { @@ -147,12 +153,14 @@ export default class RoomSubscription { handleDisconnection = () => { this.hasReconnected = true; + this.connectionCycle += 1; reduxStore.dispatch(clearUserTyping()); }; /** A new DDP handshake also means a reconnect, including reopens that emit no `close`. */ handleReconnection = () => { this.hasReconnected = true; + this.connectionCycle += 1; }; /** @@ -163,8 +171,14 @@ export default class RoomSubscription { if (!this.isAlive || !this.hasReconnected) { return; } + const cycle = this.connectionCycle; + const isStale = () => !this.isAlive || cycle !== this.connectionCycle; try { - await this.fetchMissedMessages(); + await this.fetchMissedMessages(isStale); + if (isStale()) { + // A new cycle is already under way and will fetch for itself; this result was dropped. + return; + } // Kept set until the fetch succeeds, so a failed one is retried on the next ack. this.hasReconnected = false; } catch (e) { @@ -172,8 +186,11 @@ export default class RoomSubscription { } }; - fetchMissedMessages = async () => { - await loadMissedMessages({ rid: this.rid }); + fetchMissedMessages = async (isStale: () => boolean) => { + await loadMissedMessages({ rid: this.rid, isStale }); + if (isStale()) { + return; + } this.read(); }; diff --git a/app/lib/methods/subscriptions/roomCloseCursor.test.ts b/app/lib/methods/subscriptions/roomCloseCursor.test.ts index fd1d9f3a52a..141de196368 100644 --- a/app/lib/methods/subscriptions/roomCloseCursor.test.ts +++ b/app/lib/methods/subscriptions/roomCloseCursor.test.ts @@ -115,7 +115,7 @@ describe('closing a room while offline must not advance the sync cursor', () => await new RoomSubscription(RID).unsubscribe(); await Promise.resolve(); - await new RoomSubscription(RID).fetchMissedMessages(); + await new RoomSubscription(RID).fetchMissedMessages(() => false); expect(mockedUpdateMessages).toHaveBeenCalledWith( expect.objectContaining({ From 27f291a4361341488d693e3f2427e323d10e6e2a Mon Sep 17 00:00:00 2001 From: Diego Mello Date: Wed, 5 Aug 2026 15:17:20 -0300 Subject: [PATCH 4/5] fix: recover a room whose subscription has no last open MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit On reconnect, a room with no `lastOpen` built no sync request and returned clean, so messages that arrived while the connection was down were silently and permanently absent with nothing left that could fetch them. The catch-up now delegates such a room to `loadMessagesForRoom`, which is batch-capped and emits a loader row for what it could not reach, and seeds `lastOpen` from real server timestamps so the next reconnect syncs normally. The staleness guard is threaded into that path, so a superseded connection cycle still writes nothing. Also in the same area: - the sync walk is capped at 10 pages, so a room can no longer walk unbounded history across repeated connection cycles; - servers below 7.1.0 no longer receive a request with an undefined timestamp, since the no-cursor case short-circuits before the legacy branch; - a sync response carrying no `cursor` no longer throws, which used to leave the room refetching on every stream acknowledgement; - the direct-message subscription stub no longer invents `ts`, `ls` or `roomUpdatedAt` from the device clock — `ls` anchors the unread separator, so that poisoning was already user-visible. The subscription type widens to admit their absence. --- app/definitions/ISubscription.ts | 6 +- ...reateDirectMessageSubscriptionStub.test.ts | 7 +- .../createDirectMessageSubscriptionStub.ts | 7 +- app/lib/methods/loadMessagesForRoom.test.ts | 15 ++++ app/lib/methods/loadMessagesForRoom.ts | 13 ++++ app/lib/methods/loadMissedMessages.test.ts | 73 ++++++++++++++++++- app/lib/methods/loadMissedMessages.ts | 42 +++++++++-- .../subscriptions/room.reconnectFetch.test.ts | 4 +- .../subscriptions/room.resumeSync.test.ts | 5 +- app/views/RoomView/index.tsx | 2 +- 10 files changed, 152 insertions(+), 22 deletions(-) diff --git a/app/definitions/ISubscription.ts b/app/definitions/ISubscription.ts index 6bbd836889b..b6fb99b09ac 100644 --- a/app/definitions/ISubscription.ts +++ b/app/definitions/ISubscription.ts @@ -44,8 +44,8 @@ export interface ISubscription { v?: IVisitor; f: boolean; t: SubscriptionType; // TODO: we need to review this type later - ts: string | Date; - ls: Date; + ts?: string | Date; + ls?: Date; name: string; fname?: string; sanitizedFname?: string; @@ -61,7 +61,7 @@ export interface ISubscription { tunread: string[]; tunreadUser?: string[]; tunreadGroup?: string[]; - roomUpdatedAt: Date | number; + roomUpdatedAt?: Date | number; ro: boolean; lastOpen?: Date; description?: string; diff --git a/app/lib/methods/createDirectMessageSubscriptionStub.test.ts b/app/lib/methods/createDirectMessageSubscriptionStub.test.ts index 5236ec50012..99dbe20ca39 100644 --- a/app/lib/methods/createDirectMessageSubscriptionStub.test.ts +++ b/app/lib/methods/createDirectMessageSubscriptionStub.test.ts @@ -98,9 +98,10 @@ describe('createDirectMessageSubscriptionStub', () => { expect(created.archived).toBe(false); expect(created.f).toBe(false); expect(created.ro).toBe(false); - expect(created.ts).toBeInstanceOf(Date); - expect(created.ls).toBeInstanceOf(Date); - expect(created.roomUpdatedAt).toBeInstanceOf(Date); + // No server-owned timestamp is invented from the device clock. + expect(created.ts).toBeUndefined(); + expect(created.ls).toBeUndefined(); + expect(created.roomUpdatedAt).toBeUndefined(); expect(log).not.toHaveBeenCalled(); }); diff --git a/app/lib/methods/createDirectMessageSubscriptionStub.ts b/app/lib/methods/createDirectMessageSubscriptionStub.ts index 0f6108c828f..ac2c4601109 100644 --- a/app/lib/methods/createDirectMessageSubscriptionStub.ts +++ b/app/lib/methods/createDirectMessageSubscriptionStub.ts @@ -55,7 +55,6 @@ export const createDirectMessageSubscriptionStub = async ({ const db = database.active; const subCollection = db.get(SUBSCRIPTIONS_TABLE); - const now = new Date(); await db.write(async () => { await subCollection.create((s: any) => { @@ -75,9 +74,9 @@ export const createDirectMessageSubscriptionStub = async ({ s.ro = false; s.archived = false; s.f = false; - s.ts = now; - s.ls = now; - s.roomUpdatedAt = now; + // No server-owned timestamp is invented here: `ls` anchors the unread separator and + // `ts`/`roomUpdatedAt` feed sync cursors, so a device clock value poisons them until + // the real doc arrives and overwrites them wholesale. }); }); } catch (e) { diff --git a/app/lib/methods/loadMessagesForRoom.test.ts b/app/lib/methods/loadMessagesForRoom.test.ts index 8cd7006a8a7..05215d1d451 100644 --- a/app/lib/methods/loadMessagesForRoom.test.ts +++ b/app/lib/methods/loadMessagesForRoom.test.ts @@ -170,6 +170,21 @@ describe('loadMessagesForRoom', () => { expect(mockedDispatch).not.toHaveBeenCalledWith(expect.objectContaining({ type: ROOM.HISTORY_UI_LOADER_PUSH })); }); + it('writes nothing when the connection cycle it belongs to has already ended', async () => { + const partialBatch = Array.from({ length: 5 }, (_, index) => + buildMessage({ + id: `stale-${index + 1}`, + ts: new Date(Date.UTC(2024, 0, 1, 0, 0, 5 - index)).toISOString() + }) + ); + mockedSdkGet.mockResolvedValueOnce({ success: true, messages: partialBatch } as any); + + await loadMessagesForRoom({ rid: 'ROOM_ID', t: 'c', isStale: () => true }); + + expect(mockedUpdateMessages).not.toHaveBeenCalled(); + expect(updateLastOpen).not.toHaveBeenCalled(); + }); + it('pops the ui loader when a recursive batch fetch fails after the loader was pushed', async () => { const firstBatch = buildHiddenBatch('first', 50); const networkError = new Error('boom'); diff --git a/app/lib/methods/loadMessagesForRoom.ts b/app/lib/methods/loadMessagesForRoom.ts index edb7152d914..70ac4a864be 100644 --- a/app/lib/methods/loadMessagesForRoom.ts +++ b/app/lib/methods/loadMessagesForRoom.ts @@ -119,6 +119,11 @@ export async function loadMessagesForRoom(args: { t: RoomTypes; latest?: Date; loaderItem?: TMessageModel; + /** + * Checked before the writes: a load belonging to a connection cycle that has already ended can + * resolve late and overwrite newer rows or lower `lastOpen`, so its result is dropped instead. + */ + isStale?: () => boolean; }): Promise { let uiLoaderId: string | null = null; try { @@ -128,6 +133,9 @@ export async function loadMessagesForRoom(args: { uiLoaderId = id; } }); + if (args.isStale?.()) { + return; + } if (messages?.length) { const lastMessage = messages[messages.length - 1]; const lastMessageRecord = await getMessageById(lastMessage._id as string); @@ -143,6 +151,11 @@ export async function loadMessagesForRoom(args: { await updateMessages({ rid: args.rid, update: messages, loaderItem: args.loaderItem }); } + // Re-checked because the write above awaits: the cycle can end while it runs, and `lastOpen` + // has no monotonic clamp, so a stale write would lower it. + if (args.isStale?.()) { + return; + } if (!args.latest && !args.loaderItem) { await updateLastOpen(args.rid, serverTimestamps); } diff --git a/app/lib/methods/loadMissedMessages.test.ts b/app/lib/methods/loadMissedMessages.test.ts index 8a9b9e9a9f9..0d8fbe11dd4 100644 --- a/app/lib/methods/loadMissedMessages.test.ts +++ b/app/lib/methods/loadMissedMessages.test.ts @@ -3,6 +3,7 @@ import sdk from '../services/sdk'; import updateMessages from './updateMessages'; import { getSubscriptionByRoomId } from '../database/services/Subscription'; import { updateLastOpen } from './updateLastOpen'; +import { loadMessagesForRoom } from './loadMessagesForRoom'; import { store } from '../store/auxStore'; jest.mock('../services/sdk', () => ({ @@ -24,6 +25,7 @@ jest.mock('../store/auxStore', () => ({ })); jest.mock('./updateMessages', () => jest.fn()); +jest.mock('./loadMessagesForRoom', () => ({ loadMessagesForRoom: jest.fn() })); jest.mock('./updateLastOpen', () => ({ ...jest.requireActual('./updateLastOpen'), updateLastOpen: jest.fn() @@ -34,6 +36,7 @@ const mockedSdkGet = sdk.get as jest.MockedFunction; const mockedUpdateMessages = updateMessages as jest.MockedFunction; const mockedGetSubscriptionByRoomId = getSubscriptionByRoomId as jest.MockedFunction; const mockedUpdateLastOpen = updateLastOpen as jest.MockedFunction; +const mockedLoadMessagesForRoom = loadMessagesForRoom as jest.MockedFunction; const RID = 'ROOM_ID'; @@ -65,12 +68,80 @@ describe('loadMissedMessages', () => { ); }); - it('fetches nothing when the subscription has no cursor', async () => { + it('recovers through the room history load when the subscription has no cursor', async () => { mockedGetSubscriptionByRoomId.mockResolvedValue({ lastOpen: null, t: 'p' } as never); await loadMissedMessages({ rid: RID }); + expect(mockedLoadMessagesForRoom).toHaveBeenCalledWith({ rid: RID, t: 'p' }); + // The sync walk cannot build a request without a cursor, so it must not issue one. expect(mockedSdkGet).not.toHaveBeenCalled(); + expect(mockedUpdateLastOpen).not.toHaveBeenCalled(); + }); + + // Guards the short-circuit: reaching the legacy branch without a cursor sends `lastUpdate: undefined`. + it('short-circuits the legacy branch on a server below 7.1.0 when there is no cursor', async () => { + (store.getState as jest.Mock).mockReturnValue({ server: { version: '7.0.0' } }); + mockedGetSubscriptionByRoomId.mockResolvedValue({ lastOpen: null, t: 'c' } as never); + + await loadMissedMessages({ rid: RID }); + + expect(mockedSdkGet).not.toHaveBeenCalled(); + expect(mockedLoadMessagesForRoom).toHaveBeenCalledWith({ rid: RID, t: 'c' }); + }); + + it('recovers nothing when the subscription type is not a room type', async () => { + mockedGetSubscriptionByRoomId.mockResolvedValue({ lastOpen: null, t: 'thread' } as never); + + await loadMissedMessages({ rid: RID }); + + expect(mockedLoadMessagesForRoom).not.toHaveBeenCalled(); + expect(mockedSdkGet).not.toHaveBeenCalled(); + }); + + it('passes the staleness guard into the recovery so a superseded cycle writes nothing', async () => { + mockedGetSubscriptionByRoomId.mockResolvedValue({ lastOpen: null, t: 'c' } as never); + const isStale = () => true; + + await loadMissedMessages({ rid: RID, isStale }); + + expect(mockedLoadMessagesForRoom).toHaveBeenCalledWith({ rid: RID, t: 'c', isStale }); + }); + + it('keeps a healthy cursor on the sync walk instead of delegating', async () => { + const CURSOR = new Date(Date.UTC(2024, 0, 1, 11, 0, 0)); + mockedGetSubscriptionByRoomId.mockResolvedValue({ lastOpen: CURSOR, t: 'c' } as never); + mockedSdkGet.mockResolvedValue({ result: { updated: [], deleted: [], cursor: { next: null } } } as never); + + await loadMissedMessages({ rid: RID }); + + expect(mockedLoadMessagesForRoom).not.toHaveBeenCalled(); + expect(mockedSdkGet).toHaveBeenCalledWith('chat.syncMessages', expect.objectContaining({ next: CURSOR.getTime() })); + }); + + it('does not throw when the response carries no cursor', async () => { + mockedGetSubscriptionByRoomId.mockResolvedValue({ lastOpen: new Date(Date.UTC(2024, 0, 1)), t: 'c' } as never); + mockedSdkGet.mockResolvedValue({ result: { updated: [], deleted: [] } } as never); + + await expect(loadMissedMessages({ rid: RID })).resolves.toBeUndefined(); + expect(mockedUpdateLastOpen).toHaveBeenCalledWith(RID, []); + }); + + it('stops the sync walk at the batch cap instead of paging unbounded history', async () => { + mockedGetSubscriptionByRoomId.mockResolvedValue({ lastOpen: new Date(Date.UTC(2024, 0, 1)), t: 'c' } as never); + // Every page hands back another cursor, so only the cap can end the walk. + mockedSdkGet.mockResolvedValue({ + result: { updated: [], deleted: [], cursor: { next: Date.UTC(2024, 0, 1, 11, 0, 0) } } + } as never); + + await loadMissedMessages({ rid: RID }); + for (let i = 0; i < 30; i += 1) { + await new Promise(resolve => setImmediate(resolve)); + } + + // 10 pages, each fetching an UPDATED and a DELETED request — and then it stops. + expect(mockedSdkGet).toHaveBeenCalledTimes(20); + expect(mockedUpdateLastOpen).not.toHaveBeenCalled(); }); describe('last open', () => { diff --git a/app/lib/methods/loadMissedMessages.ts b/app/lib/methods/loadMissedMessages.ts index a77920c9cc8..71ed5e008ae 100644 --- a/app/lib/methods/loadMissedMessages.ts +++ b/app/lib/methods/loadMissedMessages.ts @@ -6,9 +6,19 @@ import { store } from '../store/auxStore'; import { getSubscriptionByRoomId } from '../database/services/Subscription'; import log from './helpers/log'; import { snapshotServerTimestamps, type TServerTimestamps, updateLastOpen } from './updateLastOpen'; +import { loadMessagesForRoom } from './loadMessagesForRoom'; +import { isRoomType } from './roomTypeToApiType'; const count = 50; +/** + * The walk pages backwards with no natural floor, so an unbounded recursion can mass-insert a + * room's whole history. Hitting the cap ends the walk with the cursor unadvanced, so the next + * cycle re-walks the same pages — bounded work per cycle, at the cost of no forward progress + * for a room whose gap exceeds the cap. + */ +const MAX_PAGES = 10; + const syncMessages = async ({ roomId, next, type }: { roomId: string; next: number; type: 'UPDATED' | 'DELETED' }) => { // @ts-ignore // this method dont have type const { result } = await sdk.get('chat.syncMessages', { roomId, next, count, type }); @@ -38,20 +48,22 @@ const getSyncMessagesFromCursor = async ( const [updatedMessages, deletedMessages] = await Promise.all([updatedPromise, deletedPromise]); return { deleted: deletedMessages?.deleted ?? [], - deletedNext: deletedMessages?.cursor.next, + deletedNext: deletedMessages?.cursor?.next ?? null, updated: updatedMessages?.updated ?? [], - updatedNext: updatedMessages?.cursor.next + updatedNext: updatedMessages?.cursor?.next ?? null }; }; async function load({ rid: roomId, updatedNext, - deletedNext + deletedNext, + isStale }: { rid: string; updatedNext?: number | null; deletedNext?: number | null; + isStale?: () => boolean; }) { const sub = await getSubscriptionByRoomId(roomId); if (!sub) { @@ -59,6 +71,19 @@ async function load({ } const cursor = sub.lastOpen; + // No cursor to sync from, and no page pending: the sync walk cannot build a request at all, so + // the room's missed messages are delegated to the room history load. That path is batch-capped + // and emits a loader row for what it could not reach, and it seeds `lastOpen` from the server + // timestamps it fetched, so the next reconnect syncs from a real cursor. + // `sub.t` also carries values that are not room types ('e2e', 'thread'), which the history + // endpoints cannot be resolved from, so an unrecognised one recovers nothing. + if (!cursor && !updatedNext && !deletedNext) { + if (isRoomType(sub.t)) { + await loadMessagesForRoom({ rid: roomId, t: sub.t, isStale }); + } + return; + } + const { version: serverVersion } = store.getState().server; if (compareServerVersion(serverVersion, 'greaterThanOrEqualTo', '7.1.0')) { const result = await getSyncMessagesFromCursor(roomId, cursor?.getTime(), updatedNext, deletedNext); @@ -81,13 +106,17 @@ export async function loadMissedMessages(args: { * resolve late and overwrite newer rows or lower the cursor, so its result is dropped instead. */ isStale?: () => boolean; + /** 1-based index of the page being fetched, used to stop the walk at `MAX_PAGES`. */ + page?: number; }): Promise { + const page = args.page ?? 1; // A DELETED-only continuation fetches no UPDATED page, so it must not write the cursor again. const fetchedUpdatedPage = !!args.updatedNext || !args.deletedNext; const data = await load({ rid: args.rid, updatedNext: args.updatedNext, - deletedNext: args.deletedNext + deletedNext: args.deletedNext, + isStale: args.isStale }); if (args.isStale?.()) { return; @@ -111,13 +140,14 @@ export async function loadMissedMessages(args: { return; } - if (deletedNext || updatedNext) { + if ((deletedNext || updatedNext) && page < MAX_PAGES) { loadMissedMessages({ rid: args.rid, updatedNext, deletedNext, serverTimestamps, - isStale: args.isStale + isStale: args.isStale, + page: page + 1 }).catch(log); } diff --git a/app/lib/methods/subscriptions/room.reconnectFetch.test.ts b/app/lib/methods/subscriptions/room.reconnectFetch.test.ts index b733e7d6782..d7049fbe1c8 100644 --- a/app/lib/methods/subscriptions/room.reconnectFetch.test.ts +++ b/app/lib/methods/subscriptions/room.reconnectFetch.test.ts @@ -199,7 +199,7 @@ describe('RoomSubscription reconnect catch-up fetch', () => { expect(mockedSdkGet).not.toHaveBeenCalled(); }); - it('fetches nothing for a room without a sync cursor: RoomView owns the initial load', async () => { + it('recovers a room without a sync cursor through the room history load', async () => { await openRoom(); mockedGetSubscriptionByRoomId.mockResolvedValue({ lastOpen: null, t: 'c' } as never); @@ -207,7 +207,7 @@ describe('RoomSubscription reconnect catch-up fetch', () => { ackRoomStream(); await flush(); - expect(mockedSdkGet).not.toHaveBeenCalled(); + expect(mockedSdkGet).toHaveBeenCalledWith('channels.history', expect.objectContaining({ roomId: RID })); }); it('retries on the next ack when the fetch fails', async () => { diff --git a/app/lib/methods/subscriptions/room.resumeSync.test.ts b/app/lib/methods/subscriptions/room.resumeSync.test.ts index f4525b95409..b0699a6af94 100644 --- a/app/lib/methods/subscriptions/room.resumeSync.test.ts +++ b/app/lib/methods/subscriptions/room.resumeSync.test.ts @@ -79,12 +79,13 @@ describe('RoomSubscription resume sync', () => { ); }); - it('fetches nothing for a room without a sync cursor (null lastOpen): RoomView owns the initial load', async () => { + it('recovers a room without a sync cursor (null lastOpen) through the room history load', async () => { mockedGetSubscriptionByRoomId.mockResolvedValue({ lastOpen: null, t: 'c' } as never); await new RoomSubscription(RID).fetchMissedMessages(() => false); - expect(mockedSdkGet).not.toHaveBeenCalled(); + expect(mockedSdkGet).toHaveBeenCalledWith('channels.history', expect.objectContaining({ roomId: RID })); + expect(mockedSdkGet).not.toHaveBeenCalledWith('chat.syncMessages', expect.anything()); }); it('writes nothing to the subscription when the room is closed', async () => { diff --git a/app/views/RoomView/index.tsx b/app/views/RoomView/index.tsx index aea43ebad46..0ebfc65a76b 100644 --- a/app/views/RoomView/index.tsx +++ b/app/views/RoomView/index.tsx @@ -686,7 +686,7 @@ export class RoomView extends Component { // if room is joined if (joined && 'id' in room) { if (room.alert || room.unread || room.userMentions) { - this.setLastSeen(room.ls); + this.setLastSeen(room.ls ?? null); } else { this.setLastSeen(null); } From cc1d2965a4c08c1d5b925514cdf09d1c780557e4 Mon Sep 17 00:00:00 2001 From: Diego Mello Date: Wed, 5 Aug 2026 15:20:59 -0300 Subject: [PATCH 5/5] chore: reword comments in plainer language --- ...reateDirectMessageSubscriptionStub.test.ts | 2 +- .../createDirectMessageSubscriptionStub.ts | 6 +++--- app/lib/methods/loadMessagesForRoom.ts | 8 ++++---- app/lib/methods/loadMissedMessages.test.ts | 8 ++++---- app/lib/methods/loadMissedMessages.ts | 20 +++++++++---------- 5 files changed, 21 insertions(+), 23 deletions(-) diff --git a/app/lib/methods/createDirectMessageSubscriptionStub.test.ts b/app/lib/methods/createDirectMessageSubscriptionStub.test.ts index 99dbe20ca39..1b2d83b413b 100644 --- a/app/lib/methods/createDirectMessageSubscriptionStub.test.ts +++ b/app/lib/methods/createDirectMessageSubscriptionStub.test.ts @@ -98,7 +98,7 @@ describe('createDirectMessageSubscriptionStub', () => { expect(created.archived).toBe(false); expect(created.f).toBe(false); expect(created.ro).toBe(false); - // No server-owned timestamp is invented from the device clock. + // We don't guess timestamps from the device clock. expect(created.ts).toBeUndefined(); expect(created.ls).toBeUndefined(); expect(created.roomUpdatedAt).toBeUndefined(); diff --git a/app/lib/methods/createDirectMessageSubscriptionStub.ts b/app/lib/methods/createDirectMessageSubscriptionStub.ts index ac2c4601109..831360013aa 100644 --- a/app/lib/methods/createDirectMessageSubscriptionStub.ts +++ b/app/lib/methods/createDirectMessageSubscriptionStub.ts @@ -74,9 +74,9 @@ export const createDirectMessageSubscriptionStub = async ({ s.ro = false; s.archived = false; s.f = false; - // No server-owned timestamp is invented here: `ls` anchors the unread separator and - // `ts`/`roomUpdatedAt` feed sync cursors, so a device clock value poisons them until - // the real doc arrives and overwrites them wholesale. + // No timestamps here. They belong to the server, and we'd only be guessing from the + // device clock: `ls` places the unread separator and `ts`/`roomUpdatedAt` feed sync + // cursors, so a wrong value breaks them until the real subscription arrives. }); }); } catch (e) { diff --git a/app/lib/methods/loadMessagesForRoom.ts b/app/lib/methods/loadMessagesForRoom.ts index 70ac4a864be..90ef047fe23 100644 --- a/app/lib/methods/loadMessagesForRoom.ts +++ b/app/lib/methods/loadMessagesForRoom.ts @@ -120,8 +120,8 @@ export async function loadMessagesForRoom(args: { latest?: Date; loaderItem?: TMessageModel; /** - * Checked before the writes: a load belonging to a connection cycle that has already ended can - * resolve late and overwrite newer rows or lower `lastOpen`, so its result is dropped instead. + * Checked before we save anything. A load from a connection that already dropped can come back + * late and overwrite newer messages or push `lastOpen` backwards, so we throw its result away. */ isStale?: () => boolean; }): Promise { @@ -151,8 +151,8 @@ export async function loadMessagesForRoom(args: { await updateMessages({ rid: args.rid, update: messages, loaderItem: args.loaderItem }); } - // Re-checked because the write above awaits: the cycle can end while it runs, and `lastOpen` - // has no monotonic clamp, so a stale write would lower it. + // Checked again because the save above is async: the connection can drop while it runs, and + // nothing stops `lastOpen` from moving backwards, so a late write would lower it. if (args.isStale?.()) { return; } diff --git a/app/lib/methods/loadMissedMessages.test.ts b/app/lib/methods/loadMissedMessages.test.ts index 0d8fbe11dd4..d35b3191dd9 100644 --- a/app/lib/methods/loadMissedMessages.test.ts +++ b/app/lib/methods/loadMissedMessages.test.ts @@ -74,12 +74,12 @@ describe('loadMissedMessages', () => { await loadMissedMessages({ rid: RID }); expect(mockedLoadMessagesForRoom).toHaveBeenCalledWith({ rid: RID, t: 'p' }); - // The sync walk cannot build a request without a cursor, so it must not issue one. + // Without a cursor there's no request to make, so it must not try. expect(mockedSdkGet).not.toHaveBeenCalled(); expect(mockedUpdateLastOpen).not.toHaveBeenCalled(); }); - // Guards the short-circuit: reaching the legacy branch without a cursor sends `lastUpdate: undefined`. + // If we ever let a cursorless room reach the legacy branch, it sends `lastUpdate: undefined`. it('short-circuits the legacy branch on a server below 7.1.0 when there is no cursor', async () => { (store.getState as jest.Mock).mockReturnValue({ server: { version: '7.0.0' } }); mockedGetSubscriptionByRoomId.mockResolvedValue({ lastOpen: null, t: 'c' } as never); @@ -129,7 +129,7 @@ describe('loadMissedMessages', () => { it('stops the sync walk at the batch cap instead of paging unbounded history', async () => { mockedGetSubscriptionByRoomId.mockResolvedValue({ lastOpen: new Date(Date.UTC(2024, 0, 1)), t: 'c' } as never); - // Every page hands back another cursor, so only the cap can end the walk. + // Every page hands back another cursor, so nothing but the cap can end this. mockedSdkGet.mockResolvedValue({ result: { updated: [], deleted: [], cursor: { next: Date.UTC(2024, 0, 1, 11, 0, 0) } } } as never); @@ -139,7 +139,7 @@ describe('loadMissedMessages', () => { await new Promise(resolve => setImmediate(resolve)); } - // 10 pages, each fetching an UPDATED and a DELETED request — and then it stops. + // 10 pages, one UPDATED and one DELETED request each, and then it stops. expect(mockedSdkGet).toHaveBeenCalledTimes(20); expect(mockedUpdateLastOpen).not.toHaveBeenCalled(); }); diff --git a/app/lib/methods/loadMissedMessages.ts b/app/lib/methods/loadMissedMessages.ts index 71ed5e008ae..dace34517ca 100644 --- a/app/lib/methods/loadMissedMessages.ts +++ b/app/lib/methods/loadMissedMessages.ts @@ -12,10 +12,9 @@ import { isRoomType } from './roomTypeToApiType'; const count = 50; /** - * The walk pages backwards with no natural floor, so an unbounded recursion can mass-insert a - * room's whole history. Hitting the cap ends the walk with the cursor unadvanced, so the next - * cycle re-walks the same pages — bounded work per cycle, at the cost of no forward progress - * for a room whose gap exceeds the cap. + * Nothing stops this walk on its own, so without a cap it can pull down a room's entire history. + * When we hit the cap we stop without saving the cursor, so the next reconnect starts over from + * the same place. That means a room further behind than 10 pages never catches up here. */ const MAX_PAGES = 10; @@ -71,12 +70,11 @@ async function load({ } const cursor = sub.lastOpen; - // No cursor to sync from, and no page pending: the sync walk cannot build a request at all, so - // the room's missed messages are delegated to the room history load. That path is batch-capped - // and emits a loader row for what it could not reach, and it seeds `lastOpen` from the server - // timestamps it fetched, so the next reconnect syncs from a real cursor. - // `sub.t` also carries values that are not room types ('e2e', 'thread'), which the history - // endpoints cannot be resolved from, so an unrecognised one recovers nothing. + // Without a cursor there is nothing to sync from, so we fall back to loading the room's recent + // history instead. That load stops after a few batches and leaves a "load more" row behind for + // whatever it didn't reach, and it saves a real cursor so the next reconnect can sync normally. + // `sub.t` can also hold things that aren't rooms ('e2e', 'thread'); there's no history endpoint + // for those, so we skip them. if (!cursor && !updatedNext && !deletedNext) { if (isRoomType(sub.t)) { await loadMessagesForRoom({ rid: roomId, t: sub.t, isStale }); @@ -106,7 +104,7 @@ export async function loadMissedMessages(args: { * resolve late and overwrite newer rows or lower the cursor, so its result is dropped instead. */ isStale?: () => boolean; - /** 1-based index of the page being fetched, used to stop the walk at `MAX_PAGES`. */ + /** Which page we're on, starting at 1. Only used to stop the walk once it reaches `MAX_PAGES`. */ page?: number; }): Promise { const page = args.page ?? 1;