diff --git a/packages/core/src/bridge/apis/message.ts b/packages/core/src/bridge/apis/message.ts index d3a29e8c..5092e1f7 100644 --- a/packages/core/src/bridge/apis/message.ts +++ b/packages/core/src/bridge/apis/message.ts @@ -254,6 +254,7 @@ type FriendMessage = Extract; // ── Group-history fetch guards (the server frequency-limits / kicks abusive // pulls, so keep each request small, bound the loop, and space the sends). ── const HISTORY_MAX_COUNT = 200; // hard cap on messages returned per call +const PRIVATE_HISTORY_PAGE_SIZE = 100; const HISTORY_CHUNK_SEQ = 30; // sequence span per server request (small packet) const HISTORY_MAX_REQUESTS = 12; // bound the walk-back loop const HISTORY_MIN_GAP_MS = 300; // minimum spacing between SsoGetGroupMsg sends @@ -1543,7 +1544,7 @@ export class MessageApi { selfUin, friendUid, cursor.time, - want - collected.size, + Math.min(want - collected.size, PRIVATE_HISTORY_PAGE_SIZE), cursor.random, ), ); @@ -1564,7 +1565,7 @@ export class MessageApi { break; } - if (page.cursor.time === 0 && page.cursor.random === 0) { + if (page.isComplete || (page.cursor.time === 0 && page.cursor.random === 0)) { reachedEnd = true; break; } @@ -1572,11 +1573,6 @@ export class MessageApi { const nextCursorKey = `${page.cursor.time}:${page.cursor.random}`; if (seenCursors.has(nextCursorKey)) { - if (page.messages.length === 0) { - reachedEnd = true; - break; - } - throw new Error( `private roam history cursor did not advance from ${cursor.time}/${cursor.random}`, ); diff --git a/packages/core/tests/group-history-walkback.test.ts b/packages/core/tests/group-history-walkback.test.ts index 472fded3..9990591c 100644 --- a/packages/core/tests/group-history-walkback.test.ts +++ b/packages/core/tests/group-history-walkback.test.ts @@ -208,6 +208,56 @@ describe('MessageApi history pagination', () => { expect(result.map((message) => message.ntMsgSeq)).toEqual([63213, 63214, 63215]); }); + it('splits a 200-message private history request into bounded pages', async () => { + mockC2cRoamFetch.mockImplementation( + async (_sender, _identity, _self, _friend, time, count) => ({ + messages: Array.from({ length: count }, (_, index) => ({ + ...gm(time - index), + kind: 'friend_message', + ntMsgSeq: time - index, + })), + cursor: { time: time - count, random: 1 }, + isComplete: false, + }), + ); + const result = await runHistory( + new MessageApi(ctx).getC2cLatestHistory('u_friend', 200, 10001, 1000), + ); + expect(result).toHaveLength(200); + expect(mockC2cRoamFetch.mock.calls.map((call) => call[5])).toEqual([100, 100]); + expect(result.map((message) => message.ntMsgSeq)).toEqual( + Array.from({ length: 200 }, (_, index) => 801 + index), + ); + }); + + it('honors completion even when the last page still includes a cursor', async () => { + mockC2cRoamFetch.mockResolvedValue({ + messages: [{ ...gm(1), kind: 'friend_message', ntMsgSeq: 1 }], + cursor: { time: 100, random: 2 }, + isComplete: true, + }); + const result = await runHistory( + new MessageApi(ctx).getC2cLatestHistory('u_friend', 20, 10001, 1000), + ); + expect(result).toHaveLength(1); + expect(mockC2cRoamFetch).toHaveBeenCalledOnce(); + }); + + it('does not return partial history after a later request fails', async () => { + mockC2cRoamFetch + .mockResolvedValueOnce({ + messages: [{ ...gm(1), kind: 'friend_message', ntMsgSeq: 1 }], + cursor: { time: 100, random: 2 }, + isComplete: false, + }) + .mockRejectedValueOnce(new Error('SsoGetRoamMsg failed: result=123')); + const assertion = expect( + new MessageApi(ctx).getC2cLatestHistory('u_friend', 20, 10001, 1000), + ).rejects.toThrow('result=123'); + await vi.runAllTimersAsync(); + await assertion; + }); + it('rejects a stalled private roam cursor instead of returning a partial page', async () => { mockC2cRoamFetch.mockResolvedValue({ cursor: { time: 1700000100, random: 0 }, diff --git a/packages/proto-defs/src/get-c2c-msg.ts b/packages/proto-defs/src/get-c2c-msg.ts index 775daa68..3e04ac55 100644 --- a/packages/proto-defs/src/get-c2c-msg.ts +++ b/packages/proto-defs/src/get-c2c-msg.ts @@ -33,7 +33,10 @@ export interface SsoGetRoamMsg { } export interface SsoGetRoamMsgResponse { + result?: pb<1, uint_32>; + errorMessage?: pb<2, string>; friendUid?: pb<3, string>; + isComplete?: pb<4, bool>; timestamp?: pb<5, uint_32>; random?: pb<6, uint_32>; messages?: pb_repeated<7, PushMsgBody>; diff --git a/packages/protocol/src/msg-push/fetch-c2c-history.ts b/packages/protocol/src/msg-push/fetch-c2c-history.ts index 26c26231..86898b31 100644 --- a/packages/protocol/src/msg-push/fetch-c2c-history.ts +++ b/packages/protocol/src/msg-push/fetch-c2c-history.ts @@ -33,6 +33,7 @@ interface RawSender { export interface C2cRoamPage { messages: FriendMessage[]; cursor: { time: number; random: number }; + isComplete: boolean; } function decodeC2cMessages( @@ -166,6 +167,17 @@ export async function fetchC2cRoamMessagePage( requirePacketResponse(result, 'SsoGetRoamMsg'), ); + if (response.result === 100000301) { + log.debug('SsoGetRoamMsg reached the end of private history'); + return { messages: [], cursor: { time: 0, random: 0 }, isComplete: true }; + } + + if (response.result) { + throw new Error( + `SsoGetRoamMsg failed: result=${response.result} ${response.errorMessage ?? ''}`.trim(), + ); + } + if (response.friendUid && response.friendUid !== friendUid) { throw new Error(`SsoGetRoamMsg response friend uid mismatch: ${response.friendUid}`); } @@ -175,7 +187,7 @@ export async function fetchC2cRoamMessagePage( messages.sort((a, b) => (a.ntMsgSeq ?? 0) - (b.ntMsgSeq ?? 0) || a.time - b.time); const cursor = { time: response.timestamp ?? 0, random: response.random ?? 0 }; log.debug( - 'SsoGetRoamMsg decoded friend=%s before=%d/%d count=%d wire=%d usable=%d next=%d/%d', + 'SsoGetRoamMsg decoded friend=%s before=%d/%d count=%d wire=%d usable=%d next=%d/%d complete=%s', friendUid, beforeTime, beforeRandom, @@ -184,6 +196,7 @@ export async function fetchC2cRoamMessagePage( messages.length, cursor.time, cursor.random, + response.isComplete === true, ); - return { messages, cursor }; + return { messages, cursor, isComplete: response.isComplete === true }; } diff --git a/packages/protocol/tests/msg-push/fetch-c2c-history.test.ts b/packages/protocol/tests/msg-push/fetch-c2c-history.test.ts index 559f2a31..b0edb35f 100644 --- a/packages/protocol/tests/msg-push/fetch-c2c-history.test.ts +++ b/packages/protocol/tests/msg-push/fetch-c2c-history.test.ts @@ -23,6 +23,7 @@ import { fetchC2cMessageRange, fetchC2cRoamMessagePage, } from '../../src/msg-push'; +import { s, v } from '../oidb-services/_pb-oracle'; const identity = { findFriend: () => undefined } as unknown as IdentityService; const previousLogLevel = getLogLevel(); @@ -341,6 +342,38 @@ describe('fetchC2cMessageRange / SsoGetC2cMsg', () => { }); describe('fetchC2cRoamMessagePage / SsoGetRoamMsg', () => { + it('rejects business errors before interpreting an empty history page', async () => { + const response = Buffer.from([...v(1, 123), ...s(2, 'request rejected')]); + await expect( + fetchC2cRoamMessagePage( + { sendRawPacket: async () => okResult(response) }, + identity, + 10001, + 'u_friend', + 1700000100, + 200, + ), + ).rejects.toThrow('result=123 request rejected'); + }); + + it.each([ + { bytes: [...v(1, 100000301)], cursor: { time: 0, random: 0 } }, + { + bytes: [...s(3, 'u_friend'), ...v(4, 1), ...v(5, 1700000000), ...v(6, 77)], + cursor: { time: 1700000000, random: 77 }, + }, + ])('preserves a server-confirmed end of history', async ({ bytes, cursor }) => { + const result = await fetchC2cRoamMessagePage( + { sendRawPacket: async () => okResult(Buffer.from(bytes)) }, + identity, + 10001, + 'u_friend', + 1700000100, + 20, + ); + expect(result).toEqual({ messages: [], cursor, isComplete: true }); + }); + it('encodes the timestamp cursor and orders both sides by canonical NT sequence', async () => { let captured: { cmd: string; body: Uint8Array } | null = null; const response = protobuf_encode({