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

@@ -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(

packages/core/tests/bridge-client-message.test.ts

@@ -20,6 +20,7 @@ function deferred<T>() {
 
 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<SsoGetC2cMsgResponse>({
+                    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<PacketSender['sendPacket']>();
+        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());

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>;
 }

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<SsoGetC2cMsgResponse>(
-        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<SsoGetC2cMsgResponse>(
+            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<void>((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) {

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<SsoGetC2cMsgResponse>(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();