From bde7f2805d6d32570bf44e5ee78a4f59c02addc2 Mon Sep 17 00:00:00 2001 From: DrSmoothl <1787882683@qq.com> Date: Wed, 21 Oct 2026 17:25:13 +0800 Subject: [PATCH] fix: handle delayed client message availability Keep self-sent private messages available when their history becomes visible shortly after sending. Preserve account isolation while completing the report. Refs #528 --- packages/core/src/bridge/bridge.ts | 6 ++- .../core/tests/bridge-client-message.test.ts | 27 ++++++++++ packages/proto-defs/src/get-c2c-msg.ts | 4 +- .../src/msg-push/client-message-receipt.ts | 50 ++++++++++++++--- .../msg-push/client-message-receipt.test.ts | 53 +++++++++++++++++++ 5 files changed, 132 insertions(+), 8 deletions(-) diff --git a/packages/core/src/bridge/bridge.ts b/packages/core/src/bridge/bridge.ts index 5ea8073e..07183b88 100644 --- a/packages/core/src/bridge/bridge.ts +++ b/packages/core/src/bridge/bridge.ts @@ -257,7 +257,11 @@ export class Bridge implements BridgeInterface { pkt.seqId, ); const work = (async () => { - const event = await fetchClientMessage(current.client, this.identity, receipt); + const source: PacketSender = { + sendPacket: (cmd, body, timeout) => + requireConnection().client.sendPacket(cmd, body, timeout), + }; + const event = await fetchClientMessage(source, this.identity, receipt); requireConnection(); await this.pipeline.processEvents(pkt, [event]); runtimeLog.trace( diff --git a/packages/core/tests/bridge-client-message.test.ts b/packages/core/tests/bridge-client-message.test.ts index e184a3d0..299b6fe9 100644 --- a/packages/core/tests/bridge-client-message.test.ts +++ b/packages/core/tests/bridge-client-message.test.ts @@ -20,6 +20,7 @@ function deferred() { afterEach(async () => { await Promise.all(bridges.splice(0).map((bridge) => bridge.dispose())); + vi.useRealTimers(); vi.restoreAllMocks(); }); @@ -94,6 +95,32 @@ async function setup() { } describe('Bridge client message recovery', () => { + it('does not retry a delayed lookup on a detached or replacement connection', async () => { + const ctx = await setup(); + vi.useFakeTimers(); + ctx.sendPacket.mockResolvedValue({ + ...response(), + responseData: Buffer.from( + protobuf_encode({ + result: 100000301, + friendUid: 'u_test_peer', + }), + ), + }); + const work = ctx.bridge.onPacket(packet()); + const rejected = expect(work).rejects.toThrow('inactive connection'); + await vi.advanceTimersByTimeAsync(0); + expect(ctx.sendPacket).toHaveBeenCalledOnce(); + ctx.bridge.detachPid(1234); + const replacement = vi.fn(); + ctx.bridge.bindPid(1234, { sendPacket: replacement }); + await vi.runAllTimersAsync(); + await rejected; + expect(ctx.sendPacket).toHaveBeenCalledOnce(); + expect(replacement).not.toHaveBeenCalled(); + expect(ctx.received).not.toHaveBeenCalled(); + }); + it('dispatches confirmed content through the normal event pipeline', async () => { const ctx = await setup(); await ctx.bridge.onPacket(packet()); diff --git a/packages/proto-defs/src/get-c2c-msg.ts b/packages/proto-defs/src/get-c2c-msg.ts index d77a221e..775daa68 100644 --- a/packages/proto-defs/src/get-c2c-msg.ts +++ b/packages/proto-defs/src/get-c2c-msg.ts @@ -1,4 +1,4 @@ -import type { pb, pb_repeated, uint_32, bool } from '@snowluma/proton'; +import type { pb, pb_repeated, uint_32, int_32, bool } from '@snowluma/proton'; import type { PushMsgBody } from './message'; // trpc.msg.register_proxy.RegisterProxy.SsoGetC2cMsg — fetch private (c2c) @@ -15,6 +15,8 @@ export interface SsoGetC2cMsg { } export interface SsoGetC2cMsgResponse { + result?: pb<1, int_32>; + errorMessage?: pb<2, string>; friendUid?: pb<4, string>; messages?: pb_repeated<7, PushMsgBody>; } diff --git a/packages/protocol/src/msg-push/client-message-receipt.ts b/packages/protocol/src/msg-push/client-message-receipt.ts index 80146195..a672664f 100644 --- a/packages/protocol/src/msg-push/client-message-receipt.ts +++ b/packages/protocol/src/msg-push/client-message-receipt.ts @@ -1,6 +1,7 @@ import { CLIENT_MESSAGE_RECEIPT_CMD, type PacketInfo } from '@snowluma/common/protocol-types'; import type { PacketSender } from '@snowluma/common/packet-sender'; import { isRealUin } from '@snowluma/common/uin'; +import { createLogger } from '@snowluma/common/logger'; import type { SendMessageResponse } from '@snowluma/proto-defs/action'; import type { SsoGetC2cMsg, SsoGetC2cMsgResponse } from '@snowluma/proto-defs/get-c2c-msg'; import { protobuf_decode, protobuf_encode } from '@snowluma/proton'; @@ -13,6 +14,9 @@ import { PkgType } from './enums'; import { SSO_GET_C2C_MSG_CMD } from './fetch-c2c-history'; import { requirePacketResponse } from './packet-response'; +const log = createLogger('MsgPush.ClientMessage'); +const historyRetryDelays = [100, 250, 500, 1000, 2000] as const; + export interface ClientMessageReceipt { readonly peerUid: string; readonly clientSequence: number; @@ -91,13 +95,47 @@ export async function fetchClientMessage( startSequence: receipt.sequence, endSequence: receipt.sequence, }); - const result = await sender.sendPacket(SSO_GET_C2C_MSG_CMD, Buffer.from(request)); - const response = protobuf_decode( - requirePacketResponse(result, 'Client message lookup'), - ); + let response: SsoGetC2cMsgResponse; - if (response.friendUid && response.friendUid !== receipt.peerUid) { - throw new Error('Client message lookup returned another conversation'); + for (let attempt = 0; ; attempt++) { + const result = await sender.sendPacket(SSO_GET_C2C_MSG_CMD, Buffer.from(request), 5000); + response = protobuf_decode( + requirePacketResponse(result, 'Client message lookup'), + ); + + if (response.friendUid && response.friendUid !== receipt.peerUid) { + throw new Error('Client message lookup returned another conversation'); + } + + // A successful send can precede visibility in message storage. Only + // its explicit not-yet-stored result permits retrying this exact ID. + if ( + response.result === 100000301 && + response.friendUid === receipt.peerUid && + !response.messages?.length + ) { + const delay = historyRetryDelays[attempt]; + + if (delay === undefined) { + throw new Error( + `Client message history is still unavailable after ${attempt + 1} attempts (code=${response.result})`, + ); + } + + log.debug( + 'Confirmed client message is not yet in history: attempt=%d nextDelayMs=%d', + attempt + 1, + delay, + ); + await new Promise((resolve) => setTimeout(resolve, delay)); + continue; + } + + if ((response.result ?? 0) !== 0) { + throw new Error(`Client message lookup failed (code=${response.result})`); + } + + break; } if (response.messages?.length !== 1) { diff --git a/packages/protocol/tests/msg-push/client-message-receipt.test.ts b/packages/protocol/tests/msg-push/client-message-receipt.test.ts index 124d2607..a19c09b3 100644 --- a/packages/protocol/tests/msg-push/client-message-receipt.test.ts +++ b/packages/protocol/tests/msg-push/client-message-receipt.test.ts @@ -18,6 +18,7 @@ const time = 1_700_000_000; const identities: IdentityService[] = []; afterEach(async () => { + vi.useRealTimers(); await Promise.all(identities.splice(0).map((identity) => identity.close())); }); @@ -47,6 +48,16 @@ function message(): PushMsgBody { }; } +function lookupResponse(response: SsoGetC2cMsgResponse): SendPacketResult { + return { + success: true, + gotResponse: true, + errorCode: 0, + errorMessage: '', + responseData: Buffer.from(protobuf_encode(response)), + }; +} + function setup(messages = [message()], friendUid = peer) { const identity = IdentityService.memory(self); identities.push(identity); @@ -255,6 +266,48 @@ describe('client message receipts', () => { await expect(setup([message()], 'u_other').run()).rejects.toThrow('conversation'); }); + it('retries only the confirmed message while storage catches up', async () => { + const ctx = setup(); + vi.useFakeTimers(); + ctx.sendPacket.mockResolvedValueOnce( + lookupResponse({ result: 100000301, friendUid: peer, messages: [] }), + ); + const work = ctx.run(); + await vi.advanceTimersByTimeAsync(99); + expect(ctx.sendPacket).toHaveBeenCalledTimes(1); + await vi.advanceTimersByTimeAsync(1); + expect(await work).toMatchObject({ ntMsgSeq: 87, clientSeq: 9 }); + expect(ctx.sendPacket).toHaveBeenCalledTimes(2); + expect(ctx.sendPacket.mock.calls[1]).toEqual(ctx.sendPacket.mock.calls[0]); + }); + + it('reports a bounded failure when storage never contains the confirmed message', async () => { + const ctx = setup(); + vi.useFakeTimers(); + ctx.sendPacket.mockResolvedValue( + lookupResponse({ result: 100000301, friendUid: peer, messages: [] }), + ); + const rejected = expect(ctx.run()).rejects.toThrow('still unavailable after 6 attempts'); + await vi.runAllTimersAsync(); + await rejected; + expect(ctx.sendPacket).toHaveBeenCalledTimes(6); + }); + + it.each([ + { result: 100000302, friendUid: peer, messages: [] }, + { result: 100000301, friendUid: 'u_other', messages: [] }, + { result: 100000301, messages: [] }, + { result: 100000301, friendUid: peer, messages: [message()] }, + ] satisfies SsoGetC2cMsgResponse[])( + 'does not retry unrelated or inconsistent errors: %j', + async (response) => { + const ctx = setup(); + ctx.sendPacket.mockResolvedValue(lookupResponse(response)); + await expect(ctx.run()).rejects.toThrow(); + expect(ctx.sendPacket).toHaveBeenCalledOnce(); + }, + ); + it('preserves distinct identities for equal text', async () => { const first = setup(); const msg = message();