From 304f3e772f0a01f6ea056907ea1cea50ee622e79 Mon Sep 17 00:00:00 2001 From: DrSmoothl <1787882683@qq.com> Date: Fri, 30 Oct 2026 12:53:21 +0800 Subject: [PATCH] fix(core): preserve private history paging failures Split larger private history reads into bounded pages and honor the end of history. Surface rejected or stalled pages instead of returning misleading empty or partial results. Fixes #605 --- packages/core/src/bridge/apis/message.ts | 10 ++-- .../core/tests/group-history-walkback.test.ts | 50 +++++++++++++++++++ packages/proto-defs/src/get-c2c-msg.ts | 3 ++ .../src/msg-push/fetch-c2c-history.ts | 17 ++++++- .../tests/msg-push/fetch-c2c-history.test.ts | 33 ++++++++++++ 5 files changed, 104 insertions(+), 9 deletions(-) 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({