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/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..1b2d83b413b 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); + // We don't guess timestamps 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..831360013aa 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 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/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/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..90ef047fe23 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 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 { 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 }); } + // 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; + } 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..d35b3191dd9 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' }); + // Without a cursor there's no request to make, so it must not try. expect(mockedSdkGet).not.toHaveBeenCalled(); + expect(mockedUpdateLastOpen).not.toHaveBeenCalled(); + }); + + // 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); + + 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 nothing but the cap can end this. + 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, one UPDATED and one DELETED request each, 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 be285d1eabf..dace34517ca 100644 --- a/app/lib/methods/loadMissedMessages.ts +++ b/app/lib/methods/loadMissedMessages.ts @@ -6,9 +6,18 @@ 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; +/** + * 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; + 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 +47,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 +70,18 @@ async function load({ } const cursor = sub.lastOpen; + // 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 }); + } + return; + } + const { version: serverVersion } = store.getState().server; if (compareServerVersion(serverVersion, 'greaterThanOrEqualTo', '7.1.0')) { const result = await getSyncMessagesFromCursor(roomId, cursor?.getTime(), updatedNext, deletedNext); @@ -76,14 +99,26 @@ 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; + /** 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; // 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; + } if (data) { const { updated, @@ -97,12 +132,20 @@ export async function loadMissedMessages(args: { // @ts-ignore // TODO: remove loaderItem obligatoriness await updateMessages({ rid: args.rid, update: updated, remove: deleted }); - if (deletedNext || updatedNext) { + // 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) && page < MAX_PAGES) { loadMissedMessages({ rid: args.rid, updatedNext, deletedNext, - serverTimestamps + serverTimestamps, + 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 new file mode 100644 index 00000000000..d7049fbe1c8 --- /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('recovers a room without a sync cursor through the room history load', async () => { + await openRoom(); + mockedGetSubscriptionByRoomId.mockResolvedValue({ lastOpen: null, t: 'c' } as never); + + reconnectSocket(); + ackRoomStream(); + await flush(); + + expect(mockedSdkGet).toHaveBeenCalledWith('channels.history', expect.objectContaining({ roomId: RID })); + }); + + 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..b0699a6af94 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(() => false); expect(mockedSdkGet).toHaveBeenCalledWith( 'chat.syncMessages', @@ -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).handleConnection(); + 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/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.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..31acc4ffe96 100644 --- a/app/lib/methods/subscriptions/room.ts +++ b/app/lib/methods/subscriptions/room.ts @@ -18,29 +18,54 @@ 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; + /** 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 () => { @@ -49,9 +74,25 @@ 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.connectedListener = sdk.onStreamData('connected', this.handleConnection); - this.disconnectedListener = sdk.onStreamData('close', this.handleConnection); + // 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.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) { @@ -68,16 +109,35 @@ 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 } } reduxStore.dispatch(clearUserTyping()); + emitter.off(roomStreamReadyEvent(this.rid), this.handleStreamReadySignal); this.removeListener(this.connectedListener); 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 => { @@ -91,16 +151,49 @@ export default class RoomSubscription { } }; - handleConnection = async () => { + 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; + }; + + /** + * 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; + } + const cycle = this.connectionCycle; + const isStale = () => !this.isAlive || cycle !== this.connectionCycle; try { - reduxStore.dispatch(clearUserTyping()); - await loadMissedMessages({ rid: this.rid }); - this.read(); + 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) { log(e); } }; + fetchMissedMessages = async (isStale: () => boolean) => { + await loadMissedMessages({ rid: this.rid, isStale }); + if (isStale()) { + return; + } + 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..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).handleConnection(); + await new RoomSubscription(RID).fetchMissedMessages(() => false); expect(mockedUpdateMessages).toHaveBeenCalledWith( expect.objectContaining({ 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); } 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); }